diff --git a/olm/olm.go b/olm/olm.go index 64208eb..c30c17d 100644 --- a/olm/olm.go +++ b/olm/olm.go @@ -536,6 +536,8 @@ func (o *Olm) StartTunnel(config TunnelConfig) { o.websocket.RegisterHandler("olm/wg/peer/update", o.handleWgPeerUpdate) o.websocket.RegisterHandler("olm/wg/peer/relay", o.handleWgPeerRelay) o.websocket.RegisterHandler("olm/wg/peer/unrelay", o.handleWgPeerUnrelay) + o.websocket.RegisterHandler("olm/wg/peer/local", o.handleWgPeerLocal) + o.websocket.RegisterHandler("olm/wg/peer/unlocal", o.handleWgPeerUnlocal) // Handlers for managing remote subnets to a peer o.websocket.RegisterHandler("olm/wg/peer/data/add", o.handleWgPeerAddData) diff --git a/olm/peer.go b/olm/peer.go index 2747d7b..7c4609b 100644 --- a/olm/peer.go +++ b/olm/peer.go @@ -293,6 +293,71 @@ func (o *Olm) handleWgPeerUnrelay(msg websocket.WSMessage) { pm.UnRelayPeer(relayData.SiteId, primaryRelay) } +// handleWgPeerLocal handles the server's acknowledgement of an "olm/wg/local" message. +// olm already switched the peer to the local endpoint before sending that message (it +// doesn't wait for permission, unlike relay), so all this needs to do is stop the retry +// sender for the given chain. +func (o *Olm) handleWgPeerLocal(msg websocket.WSMessage) { + logger.Debug("Received local-peer ack message: %v", msg.Data) + + pm := o.getPeerManager() + if pm == nil { + logger.Debug("Ignoring local ack message: peerManager is nil (shutdown in progress)") + return + } + + jsonData, err := json.Marshal(msg.Data) + if err != nil { + logger.Error("Error marshaling data: %v", err) + return + } + + var localData struct { + peers.LocalPeerAckData + ChainId string `json:"chainId"` + } + if err := json.Unmarshal(jsonData, &localData); err != nil { + logger.Error("Error unmarshaling local ack data: %v", err) + return + } + + if monitor := pm.GetPeerMonitor(); monitor != nil { + monitor.CancelLocalSend(localData.ChainId) + } +} + +// handleWgPeerUnlocal handles the server's acknowledgement of an "olm/wg/unlocal" message. +// Same as handleWgPeerLocal, olm has already fallen back from the local endpoint by the time +// it sends the notification, so this just stops the retry sender. +func (o *Olm) handleWgPeerUnlocal(msg websocket.WSMessage) { + logger.Debug("Received unlocal-peer ack message: %v", msg.Data) + + pm := o.getPeerManager() + if pm == nil { + logger.Debug("Ignoring unlocal ack message: peerManager is nil (shutdown in progress)") + return + } + + jsonData, err := json.Marshal(msg.Data) + if err != nil { + logger.Error("Error marshaling data: %v", err) + return + } + + var localData struct { + peers.LocalPeerAckData + ChainId string `json:"chainId"` + } + if err := json.Unmarshal(jsonData, &localData); err != nil { + logger.Error("Error unmarshaling unlocal ack data: %v", err) + return + } + + if monitor := pm.GetPeerMonitor(); monitor != nil { + monitor.CancelLocalSend(localData.ChainId) + } +} + func (o *Olm) handleWgPeerHolepunchAddSite(msg websocket.WSMessage) { logger.Debug("Received peer-handshake message: %v", msg.Data) diff --git a/peers/monitor/monitor.go b/peers/monitor/monitor.go index 88d3edf..7c22915 100644 --- a/peers/monitor/monitor.go +++ b/peers/monitor/monitor.go @@ -80,9 +80,9 @@ type PeerMonitor struct { 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 + // Local connection sender tracking, keyed by chainId (informational messages only) + localSends map[string]func() + localSendMu sync.Mutex // Exponential backoff fields for holepunch monitor defaultHolepunchMinInterval time.Duration // Minimum interval (initial) @@ -139,7 +139,7 @@ func NewPeerMonitor(wsClient *websocket.Client, middleDev *middleDevice.MiddleDe 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()), + localSends: make(map[string]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 @@ -458,13 +458,6 @@ func (pm *PeerMonitor) RemovePeer(siteID int) { 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) { @@ -578,45 +571,73 @@ 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. +// sendLocal notifies the server that this peer switched to a local network endpoint, with +// retry keyed by chainId. This is informational (e.g. so the server can relay the information +// to newt) - olm does not wait for an acknowledgement before using the local connection, but +// it does stop retrying once the server acks via CancelLocalSend, same as relay/unrelay. func (pm *PeerMonitor) sendLocal(siteID int, endpoint string) { if pm.wsClient == nil { return } - pm.localSendMu.Lock() - if stop, ok := pm.localSendStops[siteID]; ok { - stop() - } + chainId := generateChainId() stopFunc, _ := pm.wsClient.SendMessageInterval("olm/wg/local", map[string]interface{}{ "siteId": siteID, "endpoint": endpoint, - }, 2*time.Second, 5) - pm.localSendStops[siteID] = stopFunc + "chainId": chainId, + }, 2*time.Second, 10) + + pm.localSendMu.Lock() + pm.localSends[chainId] = stopFunc pm.localSendMu.Unlock() - logger.Info("Sent local-connection message for site %d (%s)", siteID, endpoint) + logger.Info("Sent local-connection message for site %d (%s, chain %s)", siteID, endpoint, chainId) } -// sendUnLocal notifies the server that this peer fell back from its local network endpoint. +// sendUnLocal notifies the server that this peer fell back from its local network endpoint, +// with retry keyed by chainId. func (pm *PeerMonitor) sendUnLocal(siteID int) { if pm.wsClient == nil { return } - pm.localSendMu.Lock() - if stop, ok := pm.localSendStops[siteID]; ok { - stop() - } + chainId := generateChainId() stopFunc, _ := pm.wsClient.SendMessageInterval("olm/wg/unlocal", map[string]interface{}{ - "siteId": siteID, - }, 2*time.Second, 5) - pm.localSendStops[siteID] = stopFunc + "siteId": siteID, + "chainId": chainId, + }, 2*time.Second, 10) + + pm.localSendMu.Lock() + pm.localSends[chainId] = stopFunc pm.localSendMu.Unlock() - logger.Info("Sent unlocal-connection message for site %d", siteID) + logger.Info("Sent unlocal-connection message for site %d (chain %s)", siteID, chainId) +} + +// CancelLocalSend stops the interval sender for the given chainId, if one exists. +// If chainId is empty, all active local-connection senders are stopped. +func (pm *PeerMonitor) CancelLocalSend(chainId string) { + pm.localSendMu.Lock() + defer pm.localSendMu.Unlock() + + if chainId == "" { + for id, stop := range pm.localSends { + if stop != nil { + stop() + } + delete(pm.localSends, id) + } + logger.Info("Cancelled all local-connection senders") + return + } + + if stop, ok := pm.localSends[chainId]; ok { + stop() + delete(pm.localSends, chainId) + logger.Info("Cancelled local-connection sender for chain %s", chainId) + } else { + logger.Warn("CancelLocalSend: no active sender for chain %s", chainId) + } } // CancelRelaySend stops the interval sender for the given chainId, if one exists. @@ -1042,11 +1063,11 @@ func (pm *PeerMonitor) Close() { // Stop all pending local-connection senders pm.localSendMu.Lock() - for siteID, stop := range pm.localSendStops { + for chainId, stop := range pm.localSends { if stop != nil { stop() } - delete(pm.localSendStops, siteID) + delete(pm.localSends, chainId) } pm.localSendMu.Unlock() diff --git a/peers/types.go b/peers/types.go index f6b3d3b..8d4f04f 100644 --- a/peers/types.go +++ b/peers/types.go @@ -46,6 +46,13 @@ type UnRelayPeerData struct { Endpoint string `json:"endpoint"` } +// LocalPeerAckData represents the server's acknowledgement of an "olm/wg/local" or +// "olm/wg/unlocal" message. olm has already applied the local connection switch by the time +// it sends the notification, so the ack is only used to stop the retry sender. +type LocalPeerAckData struct { + SiteId int `json:"siteId"` +} + // PeerAdd represents the data needed to add remote subnets to a peer type PeerAdd struct { SiteId int `json:"siteId"`