Merge remote-tracking branch 'origin/main' into proto-ipv6-overlay

# Conflicts:
#	client/internal/peer/status.go

client/internal/peer/handshaker.go

@@ -3,6 +3,7 @@ package peer
 import (
 	"context"
 	"errors"
+	"net/netip"
 	"sync"
 	"sync/atomic"
 
@@ -40,6 +41,10 @@ type OfferAnswer struct {
 
 	// relay server address
 	RelaySrvAddress string
+	// RelaySrvIP is the IP the remote peer is connected to on its
+	// relay server. Used as a dial target if DNS for RelaySrvAddress
+	// fails. Zero value if the peer did not advertise an IP.
+	RelaySrvIP netip.Addr
 	// SessionID is the unique identifier of the session, used to discard old messages
 	SessionID *ICESessionID
 }
@@ -217,8 +222,9 @@ func (h *Handshaker) buildOfferAnswer() OfferAnswer {
 		answer.SessionID = &sid
 	}
 
-	if addr, err := h.relay.RelayInstanceAddress(); err == nil {
+	if addr, ip, err := h.relay.RelayInstanceAddress(); err == nil {
 		answer.RelaySrvAddress = addr
+		answer.RelaySrvIP = ip
 	}
 
 	return answer

client/internal/peer/notifier_test.go

@@ -8,6 +8,7 @@ import (
 type mocListener struct {
 	lastState int
 	wg        sync.WaitGroup
+	peersWg   sync.WaitGroup
 	peers     int
 }
 
@@ -33,6 +34,7 @@ func (l *mocListener) OnAddressChanged(host, addr string) {
 }
 func (l *mocListener) OnPeersListChanged(size int) {
 	l.peers = size
+	l.peersWg.Done()
 }
 
 func (l *mocListener) setWaiter() {
@@ -43,6 +45,14 @@ func (l *mocListener) wait() {
 	l.wg.Wait()
 }
 
+func (l *mocListener) setPeersWaiter() {
+	l.peersWg.Add(1)
+}
+
+func (l *mocListener) waitPeers() {
+	l.peersWg.Wait()
+}
+
 func Test_notifier_serverState(t *testing.T) {
 
 	type scenario struct {
@@ -72,11 +82,13 @@ func Test_notifier_serverState(t *testing.T) {
 func Test_notifier_SetListener(t *testing.T) {
 	listener := &mocListener{}
 	listener.setWaiter()
+	listener.setPeersWaiter()
 
 	n := newNotifier()
 	n.lastNotification = stateConnecting
 	n.setListener(listener)
 	listener.wait()
+	listener.waitPeers()
 	if listener.lastState != n.lastNotification {
 		t.Errorf("invalid state: %d, expected: %d", listener.lastState, n.lastNotification)
 	}
@@ -85,9 +97,14 @@ func Test_notifier_SetListener(t *testing.T) {
 func Test_notifier_RemoveListener(t *testing.T) {
 	listener := &mocListener{}
 	listener.setWaiter()
+	listener.setPeersWaiter()
 	n := newNotifier()
 	n.lastNotification = stateConnecting
 	n.setListener(listener)
+	// setListener replays cached state on a goroutine; wait for both the state
+	// and peers callbacks to finish so we don't race on listener.peers.
+	listener.wait()
+	listener.waitPeers()
 	n.removeListener()
 	n.peerListChanged(1)
 

client/internal/peer/signaler.go

@@ -54,19 +54,19 @@ func (s *Signaler) signalOfferAnswer(offerAnswer OfferAnswer, remoteKey string,
 			log.Warnf("failed to get session ID bytes: %v", err)
 		}
 	}
-	msg, err := signal.MarshalCredential(
-		s.wgPrivateKey,
-		offerAnswer.WgListenPort,
-		remoteKey,
-		&signal.Credential{
+	msg, err := signal.MarshalCredential(s.wgPrivateKey, remoteKey, signal.CredentialPayload{
+		Type:         bodyType,
+		WgListenPort: offerAnswer.WgListenPort,
+		Credential: &signal.Credential{
 			UFrag: offerAnswer.IceCredentials.UFrag,
 			Pwd:   offerAnswer.IceCredentials.Pwd,
 		},
-		bodyType,
-		offerAnswer.RosenpassPubKey,
-		offerAnswer.RosenpassAddr,
-		offerAnswer.RelaySrvAddress,
-		sessionIDBytes)
+		RosenpassPubKey: offerAnswer.RosenpassPubKey,
+		RosenpassAddr:   offerAnswer.RosenpassAddr,
+		RelaySrvAddress: offerAnswer.RelaySrvAddress,
+		RelaySrvIP:      offerAnswer.RelaySrvIP,
+		SessionID:       sessionIDBytes,
+	})
 	if err != nil {
 		return err
 	}

client/internal/peer/status.go

@@ -323,10 +323,10 @@ func (d *Status) RemovePeer(peerPubKey string) error {
 // UpdatePeerState updates peer status
 func (d *Status) UpdatePeerState(receivedState State) error {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	peerState, ok := d.peers[receivedState.PubKey]
 	if !ok {
+		d.mux.Unlock()
 		return errors.New("peer doesn't exist")
 	}
 
@@ -346,23 +346,29 @@ func (d *Status) UpdatePeerState(receivedState State) error {
 
 	d.peers[receivedState.PubKey] = peerState
 
-	if hasConnStatusChanged(oldState, receivedState.ConnStatus) {
-		d.notifyPeerListChanged()
-	}
-
+	notifyList := hasConnStatusChanged(oldState, receivedState.ConnStatus)
 	// when we close the connection we will not notify the router manager
-	if receivedState.ConnStatus == StatusIdle {
-		d.notifyPeerStateChangeListeners(receivedState.PubKey)
+	notifyRouter := receivedState.ConnStatus == StatusIdle
+	routerSnapshot := d.snapshotRouterPeersLocked(receivedState.PubKey, notifyRouter)
+	numPeers := d.numOfPeers()
+
+	d.mux.Unlock()
+
+	if notifyList {
+		d.notifier.peerListChanged(numPeers)
+	}
+	if notifyRouter {
+		d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
 	}
 	return nil
 }
 
 func (d *Status) AddPeerStateRoute(peer string, route string, resourceId route.ResID) error {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	peerState, ok := d.peers[peer]
 	if !ok {
+		d.mux.Unlock()
 		return errors.New("peer doesn't exist")
 	}
 
@@ -374,17 +380,20 @@ func (d *Status) AddPeerStateRoute(peer string, route string, resourceId route.R
 		d.routeIDLookup.AddRemoteRouteID(resourceId, pref)
 	}
 
+	numPeers := d.numOfPeers()
+	d.mux.Unlock()
+
 	// todo: consider to make sense of this notification or not
-	d.notifyPeerListChanged()
+	d.notifier.peerListChanged(numPeers)
 	return nil
 }
 
 func (d *Status) RemovePeerStateRoute(peer string, route string) error {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	peerState, ok := d.peers[peer]
 	if !ok {
+		d.mux.Unlock()
 		return errors.New("peer doesn't exist")
 	}
 
@@ -396,8 +405,11 @@ func (d *Status) RemovePeerStateRoute(peer string, route string) error {
 		d.routeIDLookup.RemoveRemoteRouteID(pref)
 	}
 
+	numPeers := d.numOfPeers()
+	d.mux.Unlock()
+
 	// todo: consider to make sense of this notification or not
-	d.notifyPeerListChanged()
+	d.notifier.peerListChanged(numPeers)
 	return nil
 }
 
@@ -413,10 +425,10 @@ func (d *Status) CheckRoutes(ip netip.Addr) ([]byte, bool) {
 
 func (d *Status) UpdatePeerICEState(receivedState State) error {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	peerState, ok := d.peers[receivedState.PubKey]
 	if !ok {
+		d.mux.Unlock()
 		return errors.New("peer doesn't exist")
 	}
 
@@ -434,22 +446,28 @@ func (d *Status) UpdatePeerICEState(receivedState State) error {
 
 	d.peers[receivedState.PubKey] = peerState
 
-	if hasConnStatusChanged(oldState, receivedState.ConnStatus) {
-		d.notifyPeerListChanged()
-	}
+	notifyList := hasConnStatusChanged(oldState, receivedState.ConnStatus)
+	notifyRouter := hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed)
+	routerSnapshot := d.snapshotRouterPeersLocked(receivedState.PubKey, notifyRouter)
+	numPeers := d.numOfPeers()
 
-	if hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed) {
-		d.notifyPeerStateChangeListeners(receivedState.PubKey)
+	d.mux.Unlock()
+
+	if notifyList {
+		d.notifier.peerListChanged(numPeers)
+	}
+	if notifyRouter {
+		d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
 	}
 	return nil
 }
 
 func (d *Status) UpdatePeerRelayedState(receivedState State) error {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	peerState, ok := d.peers[receivedState.PubKey]
 	if !ok {
+		d.mux.Unlock()
 		return errors.New("peer doesn't exist")
 	}
 
@@ -464,22 +482,28 @@ func (d *Status) UpdatePeerRelayedState(receivedState State) error {
 
 	d.peers[receivedState.PubKey] = peerState
 
-	if hasConnStatusChanged(oldState, receivedState.ConnStatus) {
-		d.notifyPeerListChanged()
-	}
+	notifyList := hasConnStatusChanged(oldState, receivedState.ConnStatus)
+	notifyRouter := hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed)
+	routerSnapshot := d.snapshotRouterPeersLocked(receivedState.PubKey, notifyRouter)
+	numPeers := d.numOfPeers()
 
-	if hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed) {
-		d.notifyPeerStateChangeListeners(receivedState.PubKey)
+	d.mux.Unlock()
+
+	if notifyList {
+		d.notifier.peerListChanged(numPeers)
+	}
+	if notifyRouter {
+		d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
 	}
 	return nil
 }
 
 func (d *Status) UpdatePeerRelayedStateToDisconnected(receivedState State) error {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	peerState, ok := d.peers[receivedState.PubKey]
 	if !ok {
+		d.mux.Unlock()
 		return errors.New("peer doesn't exist")
 	}
 
@@ -493,22 +517,28 @@ func (d *Status) UpdatePeerRelayedStateToDisconnected(receivedState State) error
 
 	d.peers[receivedState.PubKey] = peerState
 
-	if hasConnStatusChanged(oldState, receivedState.ConnStatus) {
-		d.notifyPeerListChanged()
-	}
+	notifyList := hasConnStatusChanged(oldState, receivedState.ConnStatus)
+	notifyRouter := hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed)
+	routerSnapshot := d.snapshotRouterPeersLocked(receivedState.PubKey, notifyRouter)
+	numPeers := d.numOfPeers()
 
-	if hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed) {
-		d.notifyPeerStateChangeListeners(receivedState.PubKey)
+	d.mux.Unlock()
+
+	if notifyList {
+		d.notifier.peerListChanged(numPeers)
+	}
+	if notifyRouter {
+		d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
 	}
 	return nil
 }
 
 func (d *Status) UpdatePeerICEStateToDisconnected(receivedState State) error {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	peerState, ok := d.peers[receivedState.PubKey]
 	if !ok {
+		d.mux.Unlock()
 		return errors.New("peer doesn't exist")
 	}
 
@@ -525,12 +555,18 @@ func (d *Status) UpdatePeerICEStateToDisconnected(receivedState State) error {
 
 	d.peers[receivedState.PubKey] = peerState
 
-	if hasConnStatusChanged(oldState, receivedState.ConnStatus) {
-		d.notifyPeerListChanged()
-	}
+	notifyList := hasConnStatusChanged(oldState, receivedState.ConnStatus)
+	notifyRouter := hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed)
+	routerSnapshot := d.snapshotRouterPeersLocked(receivedState.PubKey, notifyRouter)
+	numPeers := d.numOfPeers()
 
-	if hasStatusOrRelayedChange(oldState, receivedState.ConnStatus, oldIsRelayed, receivedState.Relayed) {
-		d.notifyPeerStateChangeListeners(receivedState.PubKey)
+	d.mux.Unlock()
+
+	if notifyList {
+		d.notifier.peerListChanged(numPeers)
+	}
+	if notifyRouter {
+		d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
 	}
 	return nil
 }
@@ -597,17 +633,33 @@ func (d *Status) UpdatePeerSSHHostKey(peerPubKey string, sshHostKey []byte) erro
 // FinishPeerListModifications this event invoke the notification
 func (d *Status) FinishPeerListModifications() {
 	d.mux.Lock()
-	defer d.mux.Unlock()
 
 	if !d.peerListChangedForNotification {
+		d.mux.Unlock()
 		return
 	}
 	d.peerListChangedForNotification = false
 
-	d.notifyPeerListChanged()
+	numPeers := d.numOfPeers()
 
+	// snapshot per-peer router state to deliver after the lock is released
+	type routerDispatch struct {
+		peerID   string
+		snapshot map[string]RouterState
+	}
+	dispatches := make([]routerDispatch, 0, len(d.peers))
 	for key := range d.peers {
-		d.notifyPeerStateChangeListeners(key)
+		snapshot := d.snapshotRouterPeersLocked(key, true)
+		if snapshot != nil {
+			dispatches = append(dispatches, routerDispatch{peerID: key, snapshot: snapshot})
+		}
+	}
+
+	d.mux.Unlock()
+
+	d.notifier.peerListChanged(numPeers)
+	for _, rd := range dispatches {
+		d.dispatchRouterPeers(rd.peerID, rd.snapshot)
 	}
 }
 
@@ -658,10 +710,15 @@ func (d *Status) GetLocalPeerState() LocalPeerState {
 // UpdateLocalPeerState updates local peer status
 func (d *Status) UpdateLocalPeerState(localPeerState LocalPeerState) {
 	d.mux.Lock()
-	defer d.mux.Unlock()
-
 	d.localPeer = localPeerState
-	d.notifyAddressChanged()
+	fqdn := d.localPeer.FQDN
+	ip := d.localPeer.IP
+	if d.localPeer.IPv6 != "" {
+		ip = ip + "\n" + d.localPeer.IPv6
+	}
+	d.mux.Unlock()
+
+	d.notifier.localAddressChanged(fqdn, ip)
 }
 
 // AddLocalPeerStateRoute adds a route to the local peer state
@@ -724,30 +781,36 @@ func (d *Status) CleanLocalPeerStateRoutes() {
 // CleanLocalPeerState cleans local peer status
 func (d *Status) CleanLocalPeerState() {
 	d.mux.Lock()
-	defer d.mux.Unlock()
-
 	d.localPeer = LocalPeerState{}
-	d.notifyAddressChanged()
+	fqdn := d.localPeer.FQDN
+	ip := d.localPeer.IP
+	d.mux.Unlock()
+
+	d.notifier.localAddressChanged(fqdn, ip)
 }
 
 // MarkManagementDisconnected sets ManagementState to disconnected
 func (d *Status) MarkManagementDisconnected(err error) {
 	d.mux.Lock()
-	defer d.mux.Unlock()
-	defer d.onConnectionChanged()
-
 	d.managementState = false
 	d.managementError = err
+	mgm := d.managementState
+	sig := d.signalState
+	d.mux.Unlock()
+
+	d.notifier.updateServerStates(mgm, sig)
 }
 
 // MarkManagementConnected sets ManagementState to connected
 func (d *Status) MarkManagementConnected() {
 	d.mux.Lock()
-	defer d.mux.Unlock()
-	defer d.onConnectionChanged()
-
 	d.managementState = true
 	d.managementError = nil
+	mgm := d.managementState
+	sig := d.signalState
+	d.mux.Unlock()
+
+	d.notifier.updateServerStates(mgm, sig)
 }
 
 // UpdateSignalAddress update the address of the signal server
@@ -781,21 +844,25 @@ func (d *Status) UpdateLazyConnection(enabled bool) {
 // MarkSignalDisconnected sets SignalState to disconnected
 func (d *Status) MarkSignalDisconnected(err error) {
 	d.mux.Lock()
-	defer d.mux.Unlock()
-	defer d.onConnectionChanged()
-
 	d.signalState = false
 	d.signalError = err
+	mgm := d.managementState
+	sig := d.signalState
+	d.mux.Unlock()
+
+	d.notifier.updateServerStates(mgm, sig)
 }
 
 // MarkSignalConnected sets SignalState to connected
 func (d *Status) MarkSignalConnected() {
 	d.mux.Lock()
-	defer d.mux.Unlock()
-	defer d.onConnectionChanged()
-
 	d.signalState = true
 	d.signalError = nil
+	mgm := d.managementState
+	sig := d.signalState
+	d.mux.Unlock()
+
+	d.notifier.updateServerStates(mgm, sig)
 }
 
 func (d *Status) UpdateRelayStates(relayResults []relay.ProbeResult) {
@@ -922,7 +989,7 @@ func (d *Status) GetRelayStates() []relay.ProbeResult {
 
 	// if the server connection is not established then we will use the general address
 	// in case of connection we will use the instance specific address
-	instanceAddr, err := d.relayMgr.RelayInstanceAddress()
+	instanceAddr, _, err := d.relayMgr.RelayInstanceAddress()
 	if err != nil {
 		// TODO add their status
 		for _, r := range d.relayMgr.ServerURLs() {
@@ -1015,18 +1082,17 @@ func (d *Status) RemoveConnectionListener() {
 	d.notifier.removeListener()
 }
 
-func (d *Status) onConnectionChanged() {
-	d.notifier.updateServerStates(d.managementState, d.signalState)
-}
-
-// notifyPeerStateChangeListeners notifies route manager about the change in peer state
-func (d *Status) notifyPeerStateChangeListeners(peerID string) {
-	subs, ok := d.changeNotify[peerID]
-	if !ok {
-		return
+// snapshotRouterPeersLocked builds the RouterState map for a peer's subscribers.
+// Caller MUST hold d.mux. Returns nil when there are no subscribers for peerID
+// or when notify is false. The snapshot is consumed later by dispatchRouterPeers
+// outside the lock so the channel send cannot stall any d.mux holder.
+func (d *Status) snapshotRouterPeersLocked(peerID string, notify bool) map[string]RouterState {
+	if !notify {
+		return nil
+	}
+	if _, ok := d.changeNotify[peerID]; !ok {
+		return nil
 	}
-
-	// collect the relevant data for router peers
 	routerPeers := make(map[string]RouterState, len(d.changeNotify))
 	for pid := range d.changeNotify {
 		s, ok := d.peers[pid]
@@ -1034,13 +1100,35 @@ func (d *Status) notifyPeerStateChangeListeners(peerID string) {
 			log.Warnf("router peer not found in peers list: %s", pid)
 			continue
 		}
-
 		routerPeers[pid] = RouterState{
 			Status:  s.ConnStatus,
 			Relayed: s.Relayed,
 			Latency: s.Latency,
 		}
 	}
+	return routerPeers
+}
+
+// dispatchRouterPeers delivers a previously snapshotted router-state map to
+// the peer's subscribers. Caller MUST NOT hold d.mux. The method takes a
+// fresh, short read of d.changeNotify under the lock to grab subscriber
+// channels, then sends outside the lock so a slow consumer cannot block other
+// d.mux holders. The send itself stays blocking (only short-circuited by the
+// subscriber's context) so peer state transitions are not silently dropped.
+func (d *Status) dispatchRouterPeers(peerID string, routerPeers map[string]RouterState) {
+	if routerPeers == nil {
+		return
+	}
+
+	d.mux.Lock()
+	subsMap, ok := d.changeNotify[peerID]
+	subs := make([]*StatusChangeSubscription, 0, len(subsMap))
+	if ok {
+		for _, sub := range subsMap {
+			subs = append(subs, sub)
+		}
+	}
+	d.mux.Unlock()
 
 	for _, sub := range subs {
 		select {
@@ -1050,18 +1138,6 @@ func (d *Status) notifyPeerStateChangeListeners(peerID string) {
 	}
 }
 
-func (d *Status) notifyPeerListChanged() {
-	d.notifier.peerListChanged(d.numOfPeers())
-}
-
-func (d *Status) notifyAddressChanged() {
-	addr := d.localPeer.IP
-	if d.localPeer.IPv6 != "" {
-		addr = addr + "\n" + d.localPeer.IPv6
-	}
-	d.notifier.localAddressChanged(d.localPeer.FQDN, addr)
-}
-
 func (d *Status) numOfPeers() int {
 	return len(d.peers) + len(d.offlinePeers)
 }

client/internal/peer/worker_relay.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"net"
+	"net/netip"
 	"sync"
 	"sync/atomic"
 
@@ -53,15 +54,19 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 	w.relaySupportedOnRemotePeer.Store(true)
 
 	// the relayManager will return with error in case if the connection has lost with relay server
-	currentRelayAddress, err := w.relayManager.RelayInstanceAddress()
+	currentRelayAddress, _, err := w.relayManager.RelayInstanceAddress()
 	if err != nil {
 		w.log.Errorf("failed to handle new offer: %s", err)
 		return
 	}
 
 	srv := w.preferredRelayServer(currentRelayAddress, remoteOfferAnswer.RelaySrvAddress)
+	var serverIP netip.Addr
+	if srv == remoteOfferAnswer.RelaySrvAddress {
+		serverIP = remoteOfferAnswer.RelaySrvIP
+	}
 
-	relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key)
+	relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key, serverIP)
 	if err != nil {
 		if errors.Is(err, relayClient.ErrConnAlreadyExists) {
 			w.log.Debugf("handled offer by reusing existing relay connection")
@@ -90,7 +95,7 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 	})
 }
 
-func (w *WorkerRelay) RelayInstanceAddress() (string, error) {
+func (w *WorkerRelay) RelayInstanceAddress() (string, netip.Addr, error) {
 	return w.relayManager.RelayInstanceAddress()
 }