rename

management/server/peer.go

@@ -1017,7 +1017,7 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy
 		}
 
 		oldHasIPv6Cap := peer.HasCapability(nbpeer.PeerCapabilityIPv6Overlay)
-		updated, versionChanged = peer.UpdateMetaIfNew(sync.Meta)
+		updated, versionChanged = peer.UpdateMetaIfNew(ctx, sync.Meta)
 		ipv6CapabilityChanged = oldHasIPv6Cap != peer.HasCapability(nbpeer.PeerCapabilityIPv6Overlay)
 		if updated {
 			am.metrics.AccountManagerMetrics().CountPeerMetUpdate()
@@ -1124,7 +1124,7 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer
 	}
 
 	var peer *nbpeer.Peer
-	var shouldStorePeer bool
+	var shouldStorePeer, shouldUpdatePeers bool
 	var peerGroupIDs []string
 
 	settings, err := am.Store.GetAccountSettings(ctx, store.LockingStrengthNone, accountID)
@@ -1151,6 +1151,7 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer
 
 			if changed {
 				shouldStorePeer = true
+				shouldUpdatePeers = true
 			}
 		}
 
@@ -1174,13 +1175,16 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer
 			}
 		}
 
+		// This is needed to keep in memory for the peer config. Otherwise browser client will end in a retry loop
+		peer.UpdateMetaIfNew(ctx, login.Meta)
+
 		return nil
 	})
 	if err != nil {
 		return nil, nil, nil, false, err
 	}
 
-	isRequiresApproval, isStatusChanged, err := am.integratedPeerValidator.IsNotValidPeer(ctx, accountID, peer, peerGroupIDs, settings.Extra)
+	isRequiresApproval, _, err := am.integratedPeerValidator.IsNotValidPeer(ctx, accountID, peer, peerGroupIDs, settings.Extra)
 	if err != nil {
 		return nil, nil, nil, false, err
 	}
@@ -1190,7 +1194,7 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer
 		return nil, nil, nil, false, err
 	}
 
-	if isStatusChanged || shouldStorePeer {
+	if shouldUpdatePeers {
 		changedPeerIDs := []string{peer.ID}
 		affectedPeerIDs := am.resolveAffectedPeersForPeerChanges(ctx, am.Store, accountID, changedPeerIDs)
 		if err = am.networkMapController.OnPeersUpdated(ctx, accountID, changedPeerIDs, affectedPeerIDs); err != nil {

management/server/peer/peer.go

@@ -1,12 +1,17 @@
 package peer
 
 import (
+	"context"
+	"fmt"
 	"net"
 	"net/netip"
 	"slices"
 	"sort"
+	"strings"
 	"time"
 
+	log "github.com/sirupsen/logrus"
+
 	"github.com/netbirdio/netbird/management/server/util"
 	"github.com/netbirdio/netbird/shared/management/http/api"
 )
@@ -162,49 +167,7 @@ type PeerSystemMeta struct { //nolint:revive
 }
 
 func (p PeerSystemMeta) isEqual(other PeerSystemMeta) bool {
-	sort.Slice(p.NetworkAddresses, func(i, j int) bool {
-		return p.NetworkAddresses[i].Mac < p.NetworkAddresses[j].Mac
-	})
-	sort.Slice(other.NetworkAddresses, func(i, j int) bool {
-		return other.NetworkAddresses[i].Mac < other.NetworkAddresses[j].Mac
-	})
-	equalNetworkAddresses := slices.EqualFunc(p.NetworkAddresses, other.NetworkAddresses, func(addr NetworkAddress, oAddr NetworkAddress) bool {
-		return addr.Mac == oAddr.Mac && addr.NetIP == oAddr.NetIP
-	})
-	if !equalNetworkAddresses {
-		return false
-	}
-
-	sort.Slice(p.Files, func(i, j int) bool {
-		return p.Files[i].Path < p.Files[j].Path
-	})
-	sort.Slice(other.Files, func(i, j int) bool {
-		return other.Files[i].Path < other.Files[j].Path
-	})
-	equalFiles := slices.EqualFunc(p.Files, other.Files, func(file File, oFile File) bool {
-		return file.Path == oFile.Path && file.Exist == oFile.Exist && file.ProcessIsRunning == oFile.ProcessIsRunning
-	})
-	if !equalFiles {
-		return false
-	}
-
-	return p.Hostname == other.Hostname &&
-		p.GoOS == other.GoOS &&
-		p.Kernel == other.Kernel &&
-		p.KernelVersion == other.KernelVersion &&
-		p.Core == other.Core &&
-		p.Platform == other.Platform &&
-		p.OS == other.OS &&
-		p.OSVersion == other.OSVersion &&
-		p.WtVersion == other.WtVersion &&
-		p.UIVersion == other.UIVersion &&
-		p.SystemSerialNumber == other.SystemSerialNumber &&
-		p.SystemProductName == other.SystemProductName &&
-		p.SystemManufacturer == other.SystemManufacturer &&
-		p.Environment.Cloud == other.Environment.Cloud &&
-		p.Environment.Platform == other.Environment.Platform &&
-		p.Flags.isEqual(other.Flags) &&
-		capabilitiesEqual(p.Capabilities, other.Capabilities)
+	return len(metaDiff(p, other)) == 0
 }
 
 func (p PeerSystemMeta) isEmpty() bool {
@@ -296,7 +259,7 @@ func (p *Peer) Copy() *Peer {
 
 // UpdateMetaIfNew updates peer's system metadata if new information is provided
 // returns true if meta was updated, false otherwise
-func (p *Peer) UpdateMetaIfNew(meta PeerSystemMeta) (updated, versionChanged bool) {
+func (p *Peer) UpdateMetaIfNew(ctx context.Context, meta PeerSystemMeta) (updated, versionChanged bool) {
 	if meta.isEmpty() {
 		return updated, versionChanged
 	}
@@ -308,14 +271,113 @@ func (p *Peer) UpdateMetaIfNew(meta PeerSystemMeta) (updated, versionChanged boo
 		meta.UIVersion = p.Meta.UIVersion
 	}
 
-	if p.Meta.isEqual(meta) {
-		return updated, versionChanged
+	oldVersion := p.Meta.WtVersion
+
+	diff := metaDiff(p.Meta, meta)
+	if len(diff) != 0 {
+		p.Meta = meta
+		updated = true
 	}
-	p.Meta = meta
-	updated = true
+
+	versionInfo := ""
+	if versionChanged {
+		versionInfo = fmt.Sprintf("version changed: %s -> %s, ", oldVersion, meta.WtVersion)
+	}
+
+	if len(diff) > 0 || versionChanged {
+		log.WithContext(ctx).
+			Debugf("peer meta updated, %s%d field(s) changed: %s", versionInfo, len(diff), strings.Join(diff, ", "))
+	}
+
 	return updated, versionChanged
 }
 
+// metaDiff returns a human-readable list of the fields that differ between the
+// old and new meta, each formatted as `field: <old> -> <new>`. It is the single
+// source of truth for meta comparison: isEqual reports equality as an empty
+// diff, so the log line can never disagree with the change decision. Slices are
+// cloned before sorting, so callers' meta is not mutated.
+func metaDiff(oldMeta, newMeta PeerSystemMeta) []string {
+	var diff []string
+	add := func(field string, oldVal, newVal any) {
+		diff = append(diff, fmt.Sprintf("%s: %v -> %v", field, oldVal, newVal))
+	}
+
+	if oldMeta.Hostname != newMeta.Hostname {
+		add("hostname", oldMeta.Hostname, newMeta.Hostname)
+	}
+	if oldMeta.GoOS != newMeta.GoOS {
+		add("goos", oldMeta.GoOS, newMeta.GoOS)
+	}
+	if oldMeta.Kernel != newMeta.Kernel {
+		add("kernel", oldMeta.Kernel, newMeta.Kernel)
+	}
+	if oldMeta.KernelVersion != newMeta.KernelVersion {
+		add("kernel_version", oldMeta.KernelVersion, newMeta.KernelVersion)
+	}
+	if oldMeta.Core != newMeta.Core {
+		add("core", oldMeta.Core, newMeta.Core)
+	}
+	if oldMeta.Platform != newMeta.Platform {
+		add("platform", oldMeta.Platform, newMeta.Platform)
+	}
+	if oldMeta.OS != newMeta.OS {
+		add("os", oldMeta.OS, newMeta.OS)
+	}
+	if oldMeta.OSVersion != newMeta.OSVersion {
+		add("os_version", oldMeta.OSVersion, newMeta.OSVersion)
+	}
+	if oldMeta.WtVersion != newMeta.WtVersion {
+		add("wt_version", oldMeta.WtVersion, newMeta.WtVersion)
+	}
+	if oldMeta.UIVersion != newMeta.UIVersion {
+		add("ui_version", oldMeta.UIVersion, newMeta.UIVersion)
+	}
+	if oldMeta.SystemSerialNumber != newMeta.SystemSerialNumber {
+		add("system_serial_number", oldMeta.SystemSerialNumber, newMeta.SystemSerialNumber)
+	}
+	if oldMeta.SystemProductName != newMeta.SystemProductName {
+		add("system_product_name", oldMeta.SystemProductName, newMeta.SystemProductName)
+	}
+	if oldMeta.SystemManufacturer != newMeta.SystemManufacturer {
+		add("system_manufacturer", oldMeta.SystemManufacturer, newMeta.SystemManufacturer)
+	}
+	if oldMeta.Environment.Cloud != newMeta.Environment.Cloud {
+		add("environment_cloud", oldMeta.Environment.Cloud, newMeta.Environment.Cloud)
+	}
+	if oldMeta.Environment.Platform != newMeta.Environment.Platform {
+		add("environment_platform", oldMeta.Environment.Platform, newMeta.Environment.Platform)
+	}
+	if !oldMeta.Flags.isEqual(newMeta.Flags) {
+		add("flags", fmt.Sprintf("%+v", oldMeta.Flags), fmt.Sprintf("%+v", newMeta.Flags))
+	}
+	if !capabilitiesEqual(oldMeta.Capabilities, newMeta.Capabilities) {
+		add("capabilities", oldMeta.Capabilities, newMeta.Capabilities)
+	}
+
+	oldAddrs := slices.Clone(oldMeta.NetworkAddresses)
+	newAddrs := slices.Clone(newMeta.NetworkAddresses)
+	sort.Slice(oldAddrs, func(i, j int) bool { return oldAddrs[i].Mac < oldAddrs[j].Mac })
+	sort.Slice(newAddrs, func(i, j int) bool { return newAddrs[i].Mac < newAddrs[j].Mac })
+	if !slices.EqualFunc(oldAddrs, newAddrs, func(a, b NetworkAddress) bool {
+		return a.Mac == b.Mac && a.NetIP == b.NetIP
+	}) {
+		add("network_addresses", fmt.Sprintf("%v", oldAddrs), fmt.Sprintf("%v", newAddrs))
+	}
+
+	oldFiles := slices.Clone(oldMeta.Files)
+	newFiles := slices.Clone(newMeta.Files)
+	sort.Slice(oldFiles, func(i, j int) bool { return oldFiles[i].Path < oldFiles[j].Path })
+	sort.Slice(newFiles, func(i, j int) bool { return newFiles[i].Path < newFiles[j].Path })
+	if !slices.EqualFunc(oldFiles, newFiles, func(a, b File) bool {
+		return a.Path == b.Path && a.Exist == b.Exist && a.ProcessIsRunning == b.ProcessIsRunning
+	}) {
+		add("files", fmt.Sprintf("%v", oldFiles), fmt.Sprintf("%v", newFiles))
+	}
+
+	return diff
+}
+
 // GetLastLogin returns the last login time of the peer.
 func (p *Peer) GetLastLogin() time.Time {
 	if p.LastLogin != nil {

signal/peer/peer.go

@@ -26,6 +26,10 @@ type Peer struct {
 
 	// a gRpc connection stream to the Peer
 	Stream proto.SignalExchange_ConnectStreamServer
+	// sendMu serializes writes to Stream. gRPC forbids concurrent SendMsg on
+	// the same ServerStream, and a peer can be the target of many senders at
+	// once.
+	sendMu sync.Mutex
 
 	// registration time
 	RegisteredAt time.Time
@@ -33,6 +37,13 @@ type Peer struct {
 	Cancel context.CancelFunc
 }
 
+// Send writes a message to the peer's stream, serializing concurrent senders.
+func (p *Peer) Send(msg *proto.EncryptedMessage) error {
+	p.sendMu.Lock()
+	defer p.sendMu.Unlock()
+	return p.Stream.Send(msg)
+}
+
 // NewPeer creates a new instance of a connected Peer
 func NewPeer(id string, stream proto.SignalExchange_ConnectStreamServer, cancel context.CancelFunc) *Peer {
 	return &Peer{

signal/server/concurrent_send_test.go

@@ -0,0 +1,67 @@
+package server
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel"
+
+	"github.com/netbirdio/netbird/shared/signal/proto"
+	"github.com/netbirdio/netbird/signal/peer"
+)
+
+// concurrencyCheckStream records the maximum number of Send calls in flight at
+// once. gRPC forbids concurrent SendMsg on the same ServerStream, so a correct
+// server must never have more than one in flight per peer.
+type concurrencyCheckStream struct {
+	proto.SignalExchange_ConnectStreamServer
+	ctx      context.Context
+	inflight atomic.Int32
+	maxSeen  atomic.Int32
+}
+
+func (s *concurrencyCheckStream) Send(*proto.EncryptedMessage) error {
+	n := s.inflight.Add(1)
+	for {
+		old := s.maxSeen.Load()
+		if n <= old || s.maxSeen.CompareAndSwap(old, n) {
+			break
+		}
+	}
+	// Widen the window so overlapping callers are reliably observed.
+	time.Sleep(time.Millisecond)
+	s.inflight.Add(-1)
+	return nil
+}
+
+func (s *concurrencyCheckStream) Context() context.Context { return s.ctx }
+
+// TestForwardMessageToPeerSerializesSend verifies that concurrent forwards to the
+// same peer never call Stream.Send concurrently, which would violate the gRPC
+// ServerStream contract.
+func TestForwardMessageToPeerSerializesSend(t *testing.T) {
+	s, err := NewServer(context.Background(), otel.Meter(""))
+	require.NoError(t, err)
+
+	const peerID = "peerX"
+	stream := &concurrencyCheckStream{ctx: context.Background()}
+	_, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	require.NoError(t, s.registry.Register(peer.NewPeer(peerID, stream, cancel)))
+
+	var wg sync.WaitGroup
+	for i := 0; i < 50; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			s.forwardMessageToPeer(context.Background(), &proto.EncryptedMessage{Key: "sender", RemoteKey: peerID})
+		}()
+	}
+	wg.Wait()
+
+	require.Equal(t, int32(1), stream.maxSeen.Load(), "Stream.Send must never run concurrently on the same peer stream")
+}

signal/server/signal.go

@@ -179,7 +179,7 @@ func (s *Server) forwardMessageToPeer(ctx context.Context, msg *proto.EncryptedM
 	sendResultChan := make(chan error, 1)
 	go func() {
 		select {
-		case sendResultChan <- dstPeer.Stream.Send(msg):
+		case sendResultChan <- dstPeer.Send(msg):
 			return
 		case <-dstPeer.Stream.Context().Done():
 			return