From 7929ce9cf900c1d5f8bfcc50e0fa7da8c34e5e5c Mon Sep 17 00:00:00 2001 From: Owen Date: Thu, 16 Jul 2026 10:57:42 -0400 Subject: [PATCH] Test local connections and choose those first --- peers/manager.go | 79 ++++++++++++- peers/monitor/monitor.go | 236 ++++++++++++++++++++++++++++++++++++++- peers/peer.go | 30 +++-- peers/types.go | 25 +++-- 4 files changed, 345 insertions(+), 25 deletions(-) diff --git a/peers/manager.go b/peers/manager.go index 8af5805..26ec4fb 100644 --- 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, diff --git a/peers/monitor/monitor.go b/peers/monitor/monitor.go index f49b397..9072d9c 100644 --- 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() diff --git a/peers/peer.go b/peers/peer.go index 7301a9c..e5d9c7c 100644 --- 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) } diff --git a/peers/types.go b/peers/types.go index 9ef1462..f6b3d3b 100644 --- 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 {