Cancel local send from chainId

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)

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)
 

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()
 

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"`