Test local connections and choose those first

2026-09-07 22:51:29 +02:00 · 2026-07-16 10:57:42 -04:00
parent 9513433c07
commit 7929ce9cf9
4 changed files with 345 additions and 25 deletions
--- a/peers/manager.go
+++ b/peers/manager.go
@@ -86,6 +86,8 @@ func NewPeerManager(config PeerManagerConfig) *PeerManager {

 	pm.optimizerTrigger = make(chan struct{}, 1)

+	pm.peerMonitor.SetLocalConnectionCallbacks(pm.LocalPeer, pm.UnLocalPeer)
+
 	return pm
 }

@@ -181,7 +183,7 @@ func (pm *PeerManager) AddPeer(siteConfig SiteConfig) error {
 	monitorAddress := strings.Split(siteConfig.ServerIP, "/")[0]
 	monitorPeer := net.JoinHostPort(monitorAddress, strconv.Itoa(int(siteConfig.ServerPort+1))) // +1 for the monitor port

-	err := pm.peerMonitor.AddPeer(siteConfig.SiteId, monitorPeer, siteConfig.Endpoint) // always use the real site endpoint for hole punch monitoring
+	err := pm.peerMonitor.AddPeer(siteConfig.SiteId, monitorPeer, siteConfig.Endpoint, siteConfig.LocalEndpoints) // always use the real site endpoint for hole punch monitoring
 	if err != nil {
 		logger.Warn("Failed to setup monitoring for site %d: %v", siteConfig.SiteId, err)
 	} else {
@@ -326,6 +328,10 @@ func (pm *PeerManager) UpdatePeer(siteConfig SiteConfig) error {
 		return fmt.Errorf("peer with site ID %d not found", siteConfig.SiteId)
 	}

+	// Preserve the currently active local endpoint (if any) across updates so an in-progress
+	// local connection isn't disrupted by an unrelated site update.
+	siteConfig.ActiveLocalEndpoint = oldPeer.ActiveLocalEndpoint
+
 	// Update aliases
 	// Remove old aliases
 	for _, alias := range oldPeer.Aliases {
@@ -473,6 +479,7 @@ func (pm *PeerManager) UpdatePeer(siteConfig SiteConfig) error {
 	}

 	pm.peerMonitor.UpdateHolepunchEndpoint(siteConfig.SiteId, siteConfig.Endpoint)
+	pm.peerMonitor.UpdateLocalEndpoints(siteConfig.SiteId, siteConfig.LocalEndpoints)

 	monitorAddress := strings.Split(siteConfig.ServerIP, "/")[0]
 	monitorPeer := net.JoinHostPort(monitorAddress, strconv.Itoa(int(siteConfig.ServerPort+1))) // +1 for the monitor port
@@ -814,6 +821,11 @@ func (pm *PeerManager) RemoveAlias(siteId int, aliasName string) error {
 func (pm *PeerManager) RelayPeer(siteId int, relayEndpoint string, relayPort uint16) {
 	pm.mu.Lock()
 	peer, exists := pm.peers[siteId]
+	if exists && peer.ActiveLocalEndpoint != "" {
+		pm.mu.Unlock()
+		logger.Info("Ignoring relay request for site %d: local connection is active", siteId)
+		return
+	}
 	if exists {
 		// Store the relay endpoint
 		peer.RelayEndpoint = relayEndpoint
@@ -926,6 +938,11 @@ func (pm *PeerManager) MarkPeerRelayed(siteID int, relayed bool) {
 func (pm *PeerManager) UnRelayPeer(siteId int, endpoint string) error {
 	pm.mu.Lock()
 	peer, exists := pm.peers[siteId]
+	if exists && peer.ActiveLocalEndpoint != "" {
+		pm.mu.Unlock()
+		logger.Info("Ignoring unrelay request for site %d: local connection is active", siteId)
+		return nil
+	}
 	if exists {
 		// Store the relay endpoint
 		peer.Endpoint = endpoint
@@ -958,6 +975,66 @@ endpoint=%s`, util.FixKey(peer.PublicKey), endpoint)
 	return nil
 }

+// LocalPeer switches a peer to a local network endpoint discovered by the peer monitor.
+// Local endpoints take priority over both the public endpoint and the relay, so this
+// bypasses relay/public-endpoint bookkeeping entirely and just updates the WireGuard
+// endpoint directly.
+func (pm *PeerManager) LocalPeer(siteId int, localEndpoint string) {
+	pm.mu.Lock()
+	peer, exists := pm.peers[siteId]
+	if exists {
+		peer.ActiveLocalEndpoint = localEndpoint
+		pm.peers[siteId] = peer
+	}
+	pm.mu.Unlock()
+
+	if !exists {
+		logger.Error("Cannot switch to local connection: peer with site ID %d not found", siteId)
+		return
+	}
+
+	// Update only the endpoint for this peer (update_only preserves other settings)
+	wgConfig := fmt.Sprintf(`public_key=%s
+update_only=true
+endpoint=%s`, util.FixKey(peer.PublicKey), localEndpoint)
+
+	if err := pm.device.IpcSet(wgConfig); err != nil {
+		logger.Error("Failed to switch peer %d to local connection: %v", siteId, err)
+		return
+	}
+
+	logger.Info("Switched peer %d to local connection at %s", siteId, localEndpoint)
+}
+
+// UnLocalPeer switches a peer away from its active local endpoint back to the public
+// endpoint, resuming the normal public/relay monitoring logic from scratch (which will
+// re-trigger relay on its own if the public endpoint also turns out to be unreachable).
+func (pm *PeerManager) UnLocalPeer(siteId int) {
+	pm.mu.Lock()
+	peer, exists := pm.peers[siteId]
+	publicDNS := pm.publicDNS
+	if exists {
+		peer.ActiveLocalEndpoint = ""
+		pm.peers[siteId] = peer
+	}
+	pm.mu.Unlock()
+
+	if !exists {
+		logger.Error("Cannot fall back from local connection: peer with site ID %d not found", siteId)
+		return
+	}
+
+	resolved, err := util.ResolveDomainUpstream(formatEndpoint(peer.Endpoint), publicDNS)
+	if err != nil {
+		logger.Error("Failed to resolve fallback endpoint for peer %d: %v", siteId, err)
+		return
+	}
+
+	if err := pm.UnRelayPeer(siteId, resolved); err != nil {
+		logger.Error("Failed to fall back peer %d from local connection: %v", siteId, err)
+	}
+}
+
 // isBetterConnection returns true if connection quality (a) is better than (b).
 // Priority: connected > disconnected, then direct > relayed, then lower RTT.
 func isBetterConnection(aConn bool, aRelay bool, aRTT time.Duration,
--- a/peers/monitor/monitor.go
+++ b/peers/monitor/monitor.go
@@ -67,6 +67,23 @@ type PeerMonitor struct {
 	holepunchMaxAttempts int          // max consecutive failures before triggering relay
 	holepunchFailures    map[int]int  // siteID -> consecutive failure count

+	// Local endpoint testing fields. Local endpoints are ip:port addresses on the
+	// site host's local network interfaces (ordered best-to-worst by the server).
+	// When one is reachable it takes priority over both the public endpoint and
+	// the relay.
+	localEndpoints      map[int][]string // siteID -> ordered candidate local endpoints
+	localActiveEndpoint map[int]string   // siteID -> currently active local endpoint ("" = not using local)
+	localFailures       map[int]int      // siteID -> consecutive failures of the active local endpoint
+	localTestTimeout    time.Duration    // timeout for each local endpoint probe
+
+	// Local connection switch callbacks, set by the PeerManager
+	localSwitchCallback   func(siteId int, endpoint string) // invoked when a local endpoint becomes active
+	localFallbackCallback func(siteId int)                  // invoked when we fall back from a local endpoint
+
+	// Local connection sender tracking, keyed by siteID (informational messages only)
+	localSendStops map[int]func()
+	localSendMu    sync.Mutex
+
 	// Exponential backoff fields for holepunch monitor
 	defaultHolepunchMinInterval time.Duration // Minimum interval (initial)
 	defaultHolepunchMaxInterval time.Duration
@@ -118,6 +135,11 @@ func NewPeerMonitor(wsClient *websocket.Client, middleDev *middleDevice.MiddleDe
 		relaySends:           make(map[string]func()),
 		holepunchMaxAttempts: 3, // Trigger relay after 3 consecutive failures
 		holepunchFailures:    make(map[int]int),
+		localEndpoints:       make(map[int][]string),
+		localActiveEndpoint:  make(map[int]string),
+		localFailures:        make(map[int]int),
+		localTestTimeout:     300 * time.Millisecond, // local network round trips should be fast
+		localSendStops:       make(map[int]func()),
 		// Rapid initial test settings: complete within ~1.5 seconds
 		rapidTestInterval:    200 * time.Millisecond, // 200ms between attempts
 		rapidTestTimeout:     400 * time.Millisecond, // 400ms timeout per attempt
@@ -235,7 +257,7 @@ func (pm *PeerMonitor) ResetPeerHolepunchInterval() {
 }

 // AddPeer adds a new peer to monitor
-func (pm *PeerMonitor) AddPeer(siteID int, endpoint string, holepunchEndpoint string) error {
+func (pm *PeerMonitor) AddPeer(siteID int, endpoint string, holepunchEndpoint string, localEndpoints []string) error {
 	pm.mutex.Lock()
 	defer pm.mutex.Unlock()

@@ -253,6 +275,9 @@ func (pm *PeerMonitor) AddPeer(siteID int, endpoint string, holepunchEndpoint st

 	pm.holepunchEndpoints[siteID] = holepunchEndpoint
 	pm.holepunchStatus[siteID] = false // Initially unknown/disconnected
+	pm.localEndpoints[siteID] = localEndpoints
+	pm.localActiveEndpoint[siteID] = ""
+	pm.localFailures[siteID] = 0

 	if pm.running {
 		if err := client.StartMonitor(func(status ConnectionStatus) {
@@ -275,6 +300,25 @@ func (pm *PeerMonitor) UpdateHolepunchEndpoint(siteID int, endpoint string) {
 	logger.Debug("Updated holepunch endpoint for site %d to %s", siteID, endpoint)
 }

+// UpdateLocalEndpoints updates the candidate local endpoints for a peer
+func (pm *PeerMonitor) UpdateLocalEndpoints(siteID int, localEndpoints []string) {
+	pm.mutex.Lock()
+	defer pm.mutex.Unlock()
+	pm.localEndpoints[siteID] = localEndpoints
+	logger.Debug("Updated local endpoints for site %d: %v", siteID, localEndpoints)
+}
+
+// SetLocalConnectionCallbacks registers the callbacks invoked when a peer switches to
+// or falls back from a local network endpoint. onLocal is called with the endpoint that
+// became active; onFallback is called when we give up on the active local endpoint and
+// resume the normal public/relay monitoring logic.
+func (pm *PeerMonitor) SetLocalConnectionCallbacks(onLocal func(siteId int, endpoint string), onFallback func(siteId int)) {
+	pm.mutex.Lock()
+	defer pm.mutex.Unlock()
+	pm.localSwitchCallback = onLocal
+	pm.localFallbackCallback = onFallback
+}
+
 // RapidTestPeer performs a rapid connectivity test for a newly added peer.
 // This is designed to quickly determine if holepunch is viable within ~1-2 seconds.
 // Returns true if the connection is viable (holepunch works), false if it should relay.
@@ -359,15 +403,25 @@ func (pm *PeerMonitor) removePeerUnlocked(siteID int) {
 // RemovePeer stops monitoring a peer and removes it from the monitor
 func (pm *PeerMonitor) RemovePeer(siteID int) {
 	pm.mutex.Lock()
-	defer pm.mutex.Unlock()

 	// remove the holepunch endpoint info
 	delete(pm.holepunchEndpoints, siteID)
 	delete(pm.holepunchStatus, siteID)
 	delete(pm.relayedPeers, siteID)
 	delete(pm.holepunchFailures, siteID)
+	delete(pm.localEndpoints, siteID)
+	delete(pm.localActiveEndpoint, siteID)
+	delete(pm.localFailures, siteID)

 	pm.removePeerUnlocked(siteID)
+	pm.mutex.Unlock()
+
+	pm.localSendMu.Lock()
+	if stop, ok := pm.localSendStops[siteID]; ok {
+		stop()
+		delete(pm.localSendStops, siteID)
+	}
+	pm.localSendMu.Unlock()
 }

 func (pm *PeerMonitor) RemoveHolepunchEndpoint(siteID int) {
@@ -481,6 +535,47 @@ func (pm *PeerMonitor) sendUnRelay(siteID int) error {
 	return nil
 }

+// sendLocal notifies the server that this peer switched to a local network endpoint.
+// This is informational only (e.g. so the server can relay the information to newt) -
+// olm does not wait for an acknowledgement before using the local connection.
+func (pm *PeerMonitor) sendLocal(siteID int, endpoint string) {
+	if pm.wsClient == nil {
+		return
+	}
+
+	pm.localSendMu.Lock()
+	if stop, ok := pm.localSendStops[siteID]; ok {
+		stop()
+	}
+	stopFunc, _ := pm.wsClient.SendMessageInterval("olm/wg/local", map[string]interface{}{
+		"siteId":   siteID,
+		"endpoint": endpoint,
+	}, 2*time.Second, 5)
+	pm.localSendStops[siteID] = stopFunc
+	pm.localSendMu.Unlock()
+
+	logger.Info("Sent local-connection message for site %d (%s)", siteID, endpoint)
+}
+
+// sendUnLocal notifies the server that this peer fell back from its local network endpoint.
+func (pm *PeerMonitor) sendUnLocal(siteID int) {
+	if pm.wsClient == nil {
+		return
+	}
+
+	pm.localSendMu.Lock()
+	if stop, ok := pm.localSendStops[siteID]; ok {
+		stop()
+	}
+	stopFunc, _ := pm.wsClient.SendMessageInterval("olm/wg/unlocal", map[string]interface{}{
+		"siteId": siteID,
+	}, 2*time.Second, 5)
+	pm.localSendStops[siteID] = stopFunc
+	pm.localSendMu.Unlock()
+
+	logger.Info("Sent unlocal-connection message for site %d", siteID)
+}
+
 // CancelRelaySend stops the interval sender for the given chainId, if one exists.
 // If chainId is empty, all active relay senders are stopped.
 func (pm *PeerMonitor) CancelRelaySend(chainId string) {
@@ -628,7 +723,8 @@ func (pm *PeerMonitor) runHolepunchMonitor() {
 			timer.Reset(currentInterval)
 			logger.Debug("Holepunch monitor interval updated, reset to %v", currentInterval)
 		case <-timer.C:
-			anyStatusChanged := pm.checkHolepunchEndpoints()
+			localChanged := pm.checkLocalEndpoints()
+			anyStatusChanged := pm.checkHolepunchEndpoints() || localChanged

 			pm.mutex.Lock()
 			if anyStatusChanged {
@@ -650,6 +746,127 @@ func (pm *PeerMonitor) runHolepunchMonitor() {
 	}
 }

+// checkLocalEndpoints tests local network endpoints for sites that have them configured.
+// For a site not currently using a local endpoint, it probes each candidate in order
+// (candidates are ordered best-to-worst by the server) and switches to the first one that
+// succeeds. For a site already using a local endpoint, it re-tests that endpoint and falls
+// back to the normal public/relay logic after a few consecutive failures.
+// Returns true if any site's local-connection status changed.
+func (pm *PeerMonitor) checkLocalEndpoints() bool {
+	pm.mutex.Lock()
+	if !pm.running {
+		pm.mutex.Unlock()
+		return false
+	}
+	if pm.holepunchTester == nil {
+		pm.mutex.Unlock()
+		return false
+	}
+	candidates := make(map[int][]string, len(pm.localEndpoints))
+	for siteID, eps := range pm.localEndpoints {
+		if len(eps) > 0 {
+			candidates[siteID] = eps
+		}
+	}
+	active := make(map[int]string, len(pm.localActiveEndpoint))
+	for siteID, ep := range pm.localActiveEndpoint {
+		active[siteID] = ep
+	}
+	timeout := pm.localTestTimeout
+	maxAttempts := pm.holepunchMaxAttempts
+	pm.mutex.Unlock()
+
+	anyChanged := false
+
+	for siteID, endpoints := range candidates {
+		if activeEndpoint := active[siteID]; activeEndpoint != "" {
+			// Already using a local endpoint - verify it's still working.
+			result := pm.holepunchTester.TestEndpoint(activeEndpoint, timeout)
+
+			pm.mutex.Lock()
+			if _, stillTracked := pm.localEndpoints[siteID]; !stillTracked {
+				pm.mutex.Unlock()
+				continue // peer was removed while we were testing
+			}
+			if result.Success {
+				pm.localFailures[siteID] = 0
+				pm.mutex.Unlock()
+				continue
+			}
+			pm.localFailures[siteID]++
+			failureCount := pm.localFailures[siteID]
+			pm.mutex.Unlock()
+
+			if failureCount >= maxAttempts {
+				logger.Warn("Local endpoint %s for site %d failed %d times, falling back to public/relay logic", activeEndpoint, siteID, failureCount)
+
+				pm.mutex.Lock()
+				pm.localActiveEndpoint[siteID] = ""
+				pm.localFailures[siteID] = 0
+				pm.holepunchFailures[siteID] = 0 // don't immediately re-trigger relay on stale failures
+				pm.mutex.Unlock()
+
+				anyChanged = true
+				pm.deactivateLocalEndpoint(siteID)
+			}
+			continue
+		}
+
+		// Not currently using a local endpoint - probe candidates in order.
+		for _, endpoint := range endpoints {
+			result := pm.holepunchTester.TestEndpoint(endpoint, timeout)
+
+			pm.mutex.Lock()
+			if _, stillTracked := pm.localEndpoints[siteID]; !stillTracked {
+				pm.mutex.Unlock()
+				break // peer was removed while we were testing
+			}
+			if !result.Success {
+				pm.mutex.Unlock()
+				continue
+			}
+			pm.localActiveEndpoint[siteID] = endpoint
+			pm.localFailures[siteID] = 0
+			pm.mutex.Unlock()
+
+			logger.Info("Local endpoint %s for site %d is reachable (RTT: %v), switching to local connection", endpoint, siteID, result.RTT)
+			anyChanged = true
+			pm.activateLocalEndpoint(siteID, endpoint)
+			break
+		}
+	}
+
+	return anyChanged
+}
+
+// activateLocalEndpoint invokes the switch callback and notifies the server that a local
+// endpoint became active for the given site.
+func (pm *PeerMonitor) activateLocalEndpoint(siteID int, endpoint string) {
+	pm.mutex.Lock()
+	cb := pm.localSwitchCallback
+	pm.mutex.Unlock()
+
+	if cb != nil {
+		cb(siteID, endpoint)
+	}
+
+	pm.sendLocal(siteID, endpoint)
+}
+
+// deactivateLocalEndpoint invokes the fallback callback and notifies the server that the
+// given site fell back from its local endpoint.
+func (pm *PeerMonitor) deactivateLocalEndpoint(siteID int) {
+	pm.mutex.Lock()
+	cb := pm.localFallbackCallback
+	pm.mutex.Unlock()
+
+	if cb != nil {
+		cb(siteID)
+	}
+
+	pm.sendUnLocal(siteID)
+}
+
 // checkHolepunchEndpoints tests all holepunch endpoints
 // Returns true if any endpoint's status changed
 func (pm *PeerMonitor) checkHolepunchEndpoints() bool {
@@ -661,6 +878,9 @@ func (pm *PeerMonitor) checkHolepunchEndpoints() bool {
 	}
 	endpoints := make(map[int]string, len(pm.holepunchEndpoints))
 	for siteID, endpoint := range pm.holepunchEndpoints {
+		if pm.localActiveEndpoint[siteID] != "" {
+			continue // using a local connection, skip public/relay monitoring
+		}
 		endpoints[siteID] = endpoint
 	}
 	timeout := pm.holepunchTimeout
@@ -777,6 +997,16 @@ func (pm *PeerMonitor) Close() {
 	}
 	pm.relaySendMu.Unlock()

+	// Stop all pending local-connection senders
+	pm.localSendMu.Lock()
+	for siteID, stop := range pm.localSendStops {
+		if stop != nil {
+			stop()
+		}
+		delete(pm.localSendStops, siteID)
+	}
+	pm.localSendMu.Unlock()
+
 	pm.mutex.Lock()
 	defer pm.mutex.Unlock()

--- a/peers/peer.go
+++ b/peers/peer.go
@@ -10,17 +10,26 @@ import (
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )

-// ConfigurePeer sets up or updates a peer within the WireGuard device
+// ConfigurePeer sets up or updates a peer within the WireGuard device.
+// If siteConfig.ActiveLocalEndpoint is set, it takes priority over both the relay and the
+// public endpoint since it's a directly-reachable address on the site host's local network.
 func ConfigurePeer(dev *device.Device, siteConfig SiteConfig, privateKey wgtypes.Key, relay bool, persistentKeepalive int, publicDNS []string) error {
-	var endpoint string
-	if relay && siteConfig.RelayEndpoint != "" {
-		endpoint = formatEndpoint(siteConfig.RelayEndpoint)
+	var siteHost string
+	if siteConfig.ActiveLocalEndpoint != "" {
+		// Local endpoints are already literal ip:port pairs on the local network, no DNS resolution needed.
+		siteHost = siteConfig.ActiveLocalEndpoint
 	} else {
-		endpoint = formatEndpoint(siteConfig.Endpoint)
-	}
-	siteHost, err := util.ResolveDomainUpstream(endpoint, publicDNS)
-	if err != nil {
-		return fmt.Errorf("failed to resolve endpoint for site %d: %v", siteConfig.SiteId, err)
+		var endpoint string
+		if relay && siteConfig.RelayEndpoint != "" {
+			endpoint = formatEndpoint(siteConfig.RelayEndpoint)
+		} else {
+			endpoint = formatEndpoint(siteConfig.Endpoint)
+		}
+		var err error
+		siteHost, err = util.ResolveDomainUpstream(endpoint, publicDNS)
+		if err != nil {
+			return fmt.Errorf("failed to resolve endpoint for site %d: %v", siteConfig.SiteId, err)
+		}
 	}

 	// Split off the CIDR of the server IP which is just a string and add /32 for the allowed IP
@@ -66,8 +75,7 @@ func ConfigurePeer(dev *device.Device, siteConfig SiteConfig, privateKey wgtypes
 	config := configBuilder.String()
 	logger.Debug("Configuring peer with config: %s", config)

-	err = dev.IpcSet(config)
-	if err != nil {
+	if err := dev.IpcSet(config); err != nil {
 		return fmt.Errorf("failed to configure WireGuard peer: %v", err)
 	}

--- a/peers/types.go
+++ b/peers/types.go
@@ -8,16 +8,21 @@ type PeerAction struct {

 // UpdatePeerData represents the data needed to update a peer
 type SiteConfig struct {
-	SiteId        int      `json:"siteId"`
-	Name          string   `json:"name,omitempty"`
-	Endpoint      string   `json:"endpoint,omitempty"`
-	RelayEndpoint string   `json:"relayEndpoint,omitempty"`
-	PublicKey     string   `json:"publicKey,omitempty"`
-	ServerIP      string   `json:"serverIP,omitempty"`
-	ServerPort    uint16   `json:"serverPort,omitempty"`
-	RemoteSubnets []string `json:"remoteSubnets,omitempty"` // optional, array of subnets that this site can access
-	AllowedIps    []string `json:"allowedIps,omitempty"`    // optional, array of allowed IPs for the peer
-	Aliases       []Alias  `json:"aliases,omitempty"`       // optional, array of alias configurations
+	SiteId         int      `json:"siteId"`
+	Name           string   `json:"name,omitempty"`
+	Endpoint       string   `json:"endpoint,omitempty"`
+	LocalEndpoints []string `json:"localEndpoints,omitempty"` // optional, ip:port endpoints on the site host's local network interfaces, ordered best-to-worst
+	RelayEndpoint  string   `json:"relayEndpoint,omitempty"`
+	PublicKey      string   `json:"publicKey,omitempty"`
+	ServerIP       string   `json:"serverIP,omitempty"`
+	ServerPort     uint16   `json:"serverPort,omitempty"`
+	RemoteSubnets  []string `json:"remoteSubnets,omitempty"` // optional, array of subnets that this site can access
+	AllowedIps     []string `json:"allowedIps,omitempty"`    // optional, array of allowed IPs for the peer
+	Aliases        []Alias  `json:"aliases,omitempty"`       // optional, array of alias configurations
+
+	// ActiveLocalEndpoint tracks the local network endpoint currently in use for this
+	// peer, if any. Not part of the wire protocol; set internally by the PeerManager.
+	ActiveLocalEndpoint string `json:"-"`
 }

 type Alias struct {