Merge pull request #4473 from openziti/link-fixes

Fix link reporting gaps found by links-test and the HA smoketest

controller/controller.go

@@ -664,20 +664,22 @@ func (c *Controller) Shutdown() {
 			}
 		}
 
-		if c.config.Db != nil {
-			if err := c.config.Db.Close(); err != nil {
-				pfxlog.Logger().WithError(err).Error("failed to close db")
-			}
-		}
-
 		go c.xweb.Shutdown()
 
+		// Raft first: its shutdown waits for the apply loop, and an apply against a closed db fails, which
+		// for a critical command halts the process.
 		if c.raftController != nil {
 			if err := c.raftController.Shutdown(); err != nil {
 				pfxlog.Logger().WithError(err).Error("failed to shutdown raft")
 			}
 		}
 
+		if c.config.Db != nil {
+			if err := c.config.Db.Close(); err != nil {
+				pfxlog.Logger().WithError(err).Error("failed to close db")
+			}
+		}
+
 		c.config.Id.StopWatchingFiles()
 		if c.config.Edge.Enrollment.SigningCert != nil {
 			c.config.Edge.Enrollment.SigningCert.StopWatchingFiles()

controller/handler_ctrl/update_link_listeners.go

@@ -52,7 +52,19 @@ func (self *updateLinkListenersHandler) HandleReceive(msg *channel.Message, ch c
 		return
 	}
 
-	if !self.router.SetLinkListeners(listeners.Listeners, listeners.Generation) {
+	previous, recorded := self.router.SetLinkListeners(listeners.Listeners, listeners.Generation)
+
+	// Routers republish on connect whenever the set may have moved since their hello, so an unchanged set
+	// is expected here and must not fan a peer-state change out to every other router. A router with no
+	// link config publishes at generation zero, which the generation check always accepts.
+	if listenersEqual(previous, listeners.Listeners) {
+		log.WithField("listenerCount", len(listeners.Listeners)).
+			WithField("generation", listeners.Generation).
+			Debug("router link listeners unchanged")
+		return
+	}
+
+	if !recorded {
 		// Superseded by an update that arrived first. Redistributing this set
 		// would push peers back onto listeners the router has already replaced.
 		log.WithField("generation", listeners.Generation).
@@ -69,3 +81,16 @@ func (self *updateLinkListenersHandler) HandleReceive(msg *channel.Message, ch c
 	// their local dial decisions.
 	self.network.RouterMessaging.RouterListenersUpdated(self.router)
 }
+
+// listenersEqual reports whether a and b hold the same listeners in the same order.
+func listenersEqual(a, b []*ctrl_pb.Listener) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if !proto.Equal(a[i], b[i]) {
+			return false
+		}
+	}
+	return true
+}

controller/handler_ctrl/update_link_listeners_test.go

@@ -0,0 +1,24 @@
+package handler_ctrl
+
+import (
+	"testing"
+
+	"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_listenersEqual(t *testing.T) {
+	listener := func(address string) *ctrl_pb.Listener {
+		return &ctrl_pb.Listener{Address: address, Protocol: "tls", Groups: []string{"default"}, LocalBinding: "default"}
+	}
+
+	req := require.New(t)
+	req.True(listenersEqual(nil, nil))
+	req.True(listenersEqual(nil, []*ctrl_pb.Listener{}))
+	req.True(listenersEqual([]*ctrl_pb.Listener{listener("tls:a:6000")}, []*ctrl_pb.Listener{listener("tls:a:6000")}))
+	req.False(listenersEqual([]*ctrl_pb.Listener{listener("tls:a:6000")}, nil))
+	req.False(listenersEqual([]*ctrl_pb.Listener{listener("tls:a:6000")}, []*ctrl_pb.Listener{listener("tls:b:6000")}))
+	req.False(listenersEqual(
+		[]*ctrl_pb.Listener{listener("tls:a:6000"), listener("tls:b:6000")},
+		[]*ctrl_pb.Listener{listener("tls:b:6000"), listener("tls:a:6000")}))
+}

controller/model/router_listeners_test.go

@@ -27,26 +27,31 @@ func listenerSet(addr string) []*ctrl_pb.Listener {
 	return []*ctrl_pb.Listener{{Address: addr, Protocol: "tls"}}
 }
 
+func setListeners(r *Router, listeners []*ctrl_pb.Listener, generation uint64) bool {
+	_, recorded := r.SetLinkListeners(listeners, generation)
+	return recorded
+}
+
 func TestRouter_SetLinkListeners_DropsSupersededGeneration(t *testing.T) {
 	req := require.New(t)
 	r := &Router{}
 
-	req.True(r.SetLinkListeners(listenerSet("a"), 1))
+	req.True(setListeners(r, listenerSet("a"), 1))
 	req.Equal("a", r.GetLinkListeners()[0].Address)
 
-	req.True(r.SetLinkListeners(listenerSet("b"), 2))
+	req.True(setListeners(r, listenerSet("b"), 2))
 	req.Equal("b", r.GetLinkListeners()[0].Address)
 
 	// An update that lost the race must not push the router back onto listeners
 	// it has already replaced, since the set is redistributed to every peer.
-	req.False(r.SetLinkListeners(listenerSet("a"), 1))
+	req.False(setListeners(r, listenerSet("a"), 1))
 	req.Equal("b", r.GetLinkListeners()[0].Address, "the newer set must survive")
 
 	// The same generation is also refused: nothing new to record.
-	req.False(r.SetLinkListeners(listenerSet("c"), 2))
+	req.False(setListeners(r, listenerSet("c"), 2))
 	req.Equal("b", r.GetLinkListeners()[0].Address)
 
-	req.True(r.SetLinkListeners(listenerSet("d"), 3))
+	req.True(setListeners(r, listenerSet("d"), 3))
 	req.Equal("d", r.GetLinkListeners()[0].Address)
 }
 
@@ -57,12 +62,12 @@ func TestRouter_ResetLinkListeners_AllowsRestartedRouter(t *testing.T) {
 	req := require.New(t)
 	r := &Router{}
 
-	req.True(r.SetLinkListeners(listenerSet("old"), 47))
-	req.False(r.SetLinkListeners(listenerSet("new"), 1), "without a reset, a restart looks superseded")
+	req.True(setListeners(r, listenerSet("old"), 47))
+	req.False(setListeners(r, listenerSet("new"), 1), "without a reset, a restart looks superseded")
 
 	r.ResetLinkListeners()
 	req.Empty(r.GetLinkListeners(), "reset clears the recorded set")
-	req.True(r.SetLinkListeners(listenerSet("new"), 1), "after a reset the new session starts fresh")
+	req.True(setListeners(r, listenerSet("new"), 1), "after a reset the new session starts fresh")
 	req.Equal("new", r.GetLinkListeners()[0].Address)
 }
 
@@ -71,6 +76,23 @@ func TestRouter_SetLinkListeners_FirstUpdateAlwaysRecorded(t *testing.T) {
 	// rather than being read as "already superseded".
 	req := require.New(t)
 	r := &Router{}
-	req.True(r.SetLinkListeners(listenerSet("a"), 0))
+	req.True(setListeners(r, listenerSet("a"), 0))
 	req.Equal("a", r.GetLinkListeners()[0].Address)
 }
+
+func TestRouter_SetLinkListeners_ReturnsTheSetItHeld(t *testing.T) {
+	req := require.New(t)
+	r := &Router{}
+
+	previous, recorded := r.SetLinkListeners(listenerSet("a"), 1)
+	req.True(recorded)
+	req.Empty(previous)
+
+	previous, recorded = r.SetLinkListeners(listenerSet("b"), 2)
+	req.True(recorded)
+	req.Equal("a", previous[0].Address, "a recorded update returns the set it replaced")
+
+	previous, recorded = r.SetLinkListeners(listenerSet("c"), 2)
+	req.False(recorded)
+	req.Equal("b", previous[0].Address, "a refused update returns the set still held")
+}

controller/model/router_model.go

@@ -167,22 +167,22 @@ func (entity *Router) addLinkListener(addr, linkProtocol string, groups []string
 	})
 }
 
-// SetLinkListeners atomically replaces the router's link listener slice.
-// Callers do not mutate the previous slice — readers may still hold and
-// iterate it safely after a Set.
-// SetLinkListeners records listeners when generation is newer than the last
-// recorded for this control channel session, and reports whether it did. Older
-// generations are dropped: a multi-underlay control channel doesn't order updates
-// across underlays, and a stale set would be redistributed to peers.
-func (entity *Router) SetLinkListeners(listeners []*ctrl_pb.Listener, generation uint64) bool {
+// SetLinkListeners records listeners when generation is newer than the one recorded for this control channel
+// session, or when the recorded generation is zero, and reports whether it did. It returns the set held before
+// the call either way, read under the same lock, so a caller can tell whether the update changed anything.
+// Older and equal generations are dropped: a multi-underlay control channel doesn't order updates across
+// underlays, and a stale set would be redistributed to peers. Slices are replaced, never mutated, so a returned
+// set stays safe to iterate.
+func (entity *Router) SetLinkListeners(listeners []*ctrl_pb.Listener, generation uint64) (previous []*ctrl_pb.Listener, recorded bool) {
 	entity.mu.Lock()
 	defer entity.mu.Unlock()
+	previous = entity.listeners
 	if generation <= entity.listenerGeneration && entity.listenerGeneration != 0 {
-		return false
+		return previous, false
 	}
 	entity.listeners = listeners
 	entity.listenerGeneration = generation
-	return true
+	return previous, true
 }
 
 // ResetLinkListeners clears the listener set and the generation it was recorded

router/handler_ctrl/peer_state.go

@@ -48,10 +48,11 @@ func (self *peerStateChangeHandler) HandleReceive(msg *channel.Message, ch chann
 		return
 	}
 
-	go self.DispatchChanges(peerStateChanges)
+	go self.DispatchChanges(ch.Id(), peerStateChanges)
 }
 
-func (self *peerStateChangeHandler) DispatchChanges(changes *ctrl_pb.PeerStateChanges) {
+// DispatchChanges applies peer state changes reported by controller ctrlId to the link registry.
+func (self *peerStateChangeHandler) DispatchChanges(ctrlId string, changes *ctrl_pb.PeerStateChanges) {
 	log := pfxlog.Logger()
 	for _, peerStateChange := range changes.Changes {
 		if peerStateChange.Id == self.env.GetRouterId().Token {
@@ -59,7 +60,7 @@ func (self *peerStateChangeHandler) DispatchChanges(changes *ctrl_pb.PeerStateCh
 		} else if peerStateChange.State == ctrl_pb.PeerState_Removed {
 			self.env.GetXlinkRegistry().RemoveLinkDest(peerStateChange.Id)
 		} else {
-			self.env.GetXlinkRegistry().UpdateLinkDest(peerStateChange.Id, peerStateChange.Version, peerStateChange.State == ctrl_pb.PeerState_Healthy, peerStateChange.Listeners)
+			self.env.GetXlinkRegistry().UpdateLinkDest(ctrlId, peerStateChange.Id, peerStateChange.Version, peerStateChange.State == ctrl_pb.PeerState_Healthy, peerStateChange.Listeners)
 		}
 	}
 }

router/link/ctrl_sync.go

@@ -0,0 +1,207 @@
+/*
+	Copyright NetFoundry Inc.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+	https://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+package link
+
+import (
+	"sync"
+
+	"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
+	"github.com/openziti/ziti/v2/router/xlink"
+)
+
+// ctrlSynchronizer tracks what the router still owes each controller. A controller it has (re)connected to owes a
+// full link refresh, and until that refresh is written the controller is not synced: incremental link
+// reports to it are held, since a report that lands ahead of the refresh is either pruned by it or re-adds
+// a link the refresh omitted. Each controller also owes the current link listener set whenever that set
+// has changed since it was last written there. Safe for concurrent use.
+type ctrlSynchronizer struct {
+	lock sync.Mutex
+	// listenersGen counts listener set changes. It starts at 1 so a controller that has never been sent
+	// the set (sentGen 0) is pending.
+	listenersGen uint64
+	states       map[string]*ctrlSyncState
+}
+
+type ctrlSyncState struct {
+	refreshPending bool
+	// refreshInFlight is true while a pool task is sending this controller's refresh, so the loop queues
+	// at most one at a time.
+	refreshInFlight bool
+	// reconnectGen counts reconnects. A refresh is built for the generation current when it was begun and
+	// only satisfies the debt if no reconnect happened before it was written.
+	reconnectGen     uint64
+	listenersSentGen uint64
+}
+
+func newCtrlSynchronizer() *ctrlSynchronizer {
+	return &ctrlSynchronizer{listenersGen: 1, states: map[string]*ctrlSyncState{}}
+}
+
+func (self *ctrlSynchronizer) state(ctrlId string) *ctrlSyncState {
+	result, found := self.states[ctrlId]
+	if !found {
+		result = &ctrlSyncState{}
+		self.states[ctrlId] = result
+	}
+	return result
+}
+
+// markReconnected records that ctrlId owes a full refresh and the current listener set.
+func (self *ctrlSynchronizer) markReconnected(ctrlId string) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	state := self.state(ctrlId)
+	state.refreshPending = true
+	state.listenersSentGen = 0
+	state.reconnectGen++
+}
+
+// beginRefresh claims the refresh owed to ctrlId, returning the reconnect generation it is being built for.
+// It reports false when no refresh is owed or one is already in flight.
+func (self *ctrlSynchronizer) beginRefresh(ctrlId string) (gen uint64, ok bool) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	state, found := self.states[ctrlId]
+	if !found || !state.refreshPending || state.refreshInFlight {
+		return 0, false
+	}
+	state.refreshInFlight = true
+	return state.reconnectGen, true
+}
+
+// markRefreshWritten records that the refresh begun for gen was written to ctrlId. It reports false, leaving
+// the refresh owed, when a reconnect happened after gen was taken: that refresh described a channel state
+// since replaced, so the controller is not synced by it.
+func (self *ctrlSynchronizer) markRefreshWritten(ctrlId string, gen uint64) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	state := self.state(ctrlId)
+	state.refreshInFlight = false
+	if state.reconnectGen != gen {
+		return false
+	}
+	state.refreshPending = false
+	return true
+}
+
+// markRefreshFailed records that the refresh in flight for ctrlId was not written. The debt stays.
+func (self *ctrlSynchronizer) markRefreshFailed(ctrlId string) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.state(ctrlId).refreshInFlight = false
+}
+
+// isRefreshInFlight reports whether a refresh for ctrlId is being sent.
+func (self *ctrlSynchronizer) isRefreshInFlight(ctrlId string) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	state, found := self.states[ctrlId]
+	return found && state.refreshInFlight
+}
+
+// isSynced reports whether incremental link reports may be sent to ctrlId. A controller never seen by
+// markReconnected is synced, which keeps the pre-reconnect-tracking behavior for anything not routed through it.
+func (self *ctrlSynchronizer) isSynced(ctrlId string) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	state, found := self.states[ctrlId]
+	return !found || !state.refreshPending
+}
+
+// markListenersChanged records that the listener set changed, making it pending for every controller. It
+// returns the new generation.
+func (self *ctrlSynchronizer) markListenersChanged() uint64 {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.listenersGen++
+	return self.listenersGen
+}
+
+// currentListenersGen returns the generation a send built now should report with markListenersSent.
+func (self *ctrlSynchronizer) currentListenersGen() uint64 {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.listenersGen
+}
+
+// isListenersSendPending reports whether ctrlId has not been sent the current listener set.
+func (self *ctrlSynchronizer) isListenersSendPending(ctrlId string) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	state, found := self.states[ctrlId]
+	return !found || state.listenersSentGen != self.listenersGen
+}
+
+// reconnectGen returns ctrlId's current reconnect generation, for fencing a send begun now against a later
+// reconnect.
+func (self *ctrlSynchronizer) reconnectGen(ctrlId string) uint64 {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.state(ctrlId).reconnectGen
+}
+
+// markListenersSent records that the listener set of generation gen was written to ctrlId over the connection
+// of reconnect generation reconnectGen. A set that changed again since gen was built stays pending, and so
+// does the whole set when ctrlId has reconnected since reconnectGen was taken: that write reached the previous
+// connection, not the current one.
+func (self *ctrlSynchronizer) markListenersSent(ctrlId string, gen uint64, reconnectGen uint64) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	state := self.state(ctrlId)
+	if state.reconnectGen != reconnectGen {
+		return
+	}
+	if gen > state.listenersSentGen {
+		state.listenersSentGen = gen
+	}
+}
+
+// forget drops tracking for controllers that are no longer in known and are owed nothing. A state that owes a
+// refresh or the listener set is kept even when its controller is absent from known: known is a snapshot, and
+// a controller that connected after it was taken has just recorded that debt. Dropping the state would read
+// as synced and nothing would ever announce to it. A settled state is safe to drop, since a controller that
+// reappears records a fresh debt.
+func (self *ctrlSynchronizer) forget(known map[string]struct{}) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	for ctrlId, state := range self.states {
+		if _, ok := known[ctrlId]; ok {
+			continue
+		}
+		settled := !state.refreshPending && !state.refreshInFlight && state.listenersSentGen == self.listenersGen
+		if settled {
+			delete(self.states, ctrlId)
+		}
+	}
+}
+
+// ListenersToProto converts the router's link listeners, stamped with the config generation they belong to,
+// to the wire shape used in the hello ListenersHeader and in UpdateLinkListeners, in the order given. The
+// generation and listeners must come from one snapshot, or a controller comparing generations may keep a stale
+// set and reject the newer one.
+func ListenersToProto(generation uint64, listeners []xlink.Listener) *ctrl_pb.Listeners {
+	result := &ctrl_pb.Listeners{Generation: generation}
+	for _, listener := range listeners {
+		result.Listeners = append(result.Listeners, &ctrl_pb.Listener{
+			Address:      listener.GetAdvertisement(),
+			Protocol:     listener.GetLinkProtocol(),
+			Groups:       listener.GetGroups(),
+			LocalBinding: listener.GetLocalBinding(),
+		})
+	}
+	return result
+}

router/link/ctrl_sync_test.go

@@ -0,0 +1,119 @@
+package link
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_ctrlSynchronizer(t *testing.T) {
+	t.Run("an unknown controller is synced, owes no refresh, and owes the listener set", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+		require.True(t, sync.isSynced("ctrl1"))
+		_, ok := sync.beginRefresh("ctrl1")
+		require.False(t, ok)
+		require.True(t, sync.isListenersSendPending("ctrl1"))
+	})
+
+	t.Run("a reconnect owes a refresh, claimed once, and sync follows the write", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+		sync.markReconnected("ctrl1")
+		require.False(t, sync.isSynced("ctrl1"))
+
+		gen, ok := sync.beginRefresh("ctrl1")
+		require.True(t, ok)
+		_, again := sync.beginRefresh("ctrl1")
+		require.False(t, again, "one refresh in flight at a time")
+
+		require.True(t, sync.markRefreshWritten("ctrl1", gen))
+		require.True(t, sync.isSynced("ctrl1"))
+		_, owed := sync.beginRefresh("ctrl1")
+		require.False(t, owed, "nothing owed after the write")
+	})
+
+	t.Run("a failed refresh releases the claim and leaves the debt", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+		sync.markReconnected("ctrl1")
+		_, ok := sync.beginRefresh("ctrl1")
+		require.True(t, ok)
+		sync.markRefreshFailed("ctrl1")
+		require.False(t, sync.isRefreshInFlight("ctrl1"))
+		_, ok = sync.beginRefresh("ctrl1")
+		require.True(t, ok)
+	})
+
+	t.Run("a reconnect during a refresh invalidates it", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+		sync.markReconnected("ctrl1")
+		gen, _ := sync.beginRefresh("ctrl1")
+		sync.markReconnected("ctrl1")
+		require.False(t, sync.markRefreshWritten("ctrl1", gen), "a refresh built before the later reconnect does not satisfy it")
+		require.False(t, sync.isSynced("ctrl1"))
+		gen2, ok := sync.beginRefresh("ctrl1")
+		require.True(t, ok)
+		require.True(t, sync.markRefreshWritten("ctrl1", gen2))
+		require.True(t, sync.isSynced("ctrl1"))
+	})
+
+	t.Run("listeners sent for the current generation clears the debt, an older generation does not", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+		gen := sync.currentListenersGen()
+		sync.markListenersChanged()
+		sync.markListenersSent("ctrl1", gen, sync.reconnectGen("ctrl1"))
+		require.True(t, sync.isListenersSendPending("ctrl1"))
+		sync.markListenersSent("ctrl1", sync.currentListenersGen(), sync.reconnectGen("ctrl1"))
+		require.False(t, sync.isListenersSendPending("ctrl1"))
+	})
+
+	t.Run("a reconnect makes the listener set owed again", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+		sync.markListenersSent("ctrl1", sync.currentListenersGen(), sync.reconnectGen("ctrl1"))
+		require.False(t, sync.isListenersSendPending("ctrl1"))
+		sync.markReconnected("ctrl1")
+		require.True(t, sync.isListenersSendPending("ctrl1"))
+	})
+
+	t.Run("a listener send written before a reconnect does not satisfy the new connection", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+		sync.markReconnected("ctrl1")
+		gen, _ := sync.beginRefresh("ctrl1")
+		require.True(t, sync.markRefreshWritten("ctrl1", gen))
+
+		listenersGen := sync.currentListenersGen()
+		staleReconnectGen := sync.reconnectGen("ctrl1")
+
+		sync.markReconnected("ctrl1") // reconnect after the write, before its completion
+		sync.markListenersSent("ctrl1", listenersGen, staleReconnectGen)
+		require.True(t, sync.isListenersSendPending("ctrl1"), "the stale completion must not mark the new connection sent")
+
+		sync.markListenersSent("ctrl1", listenersGen, sync.reconnectGen("ctrl1"))
+		require.False(t, sync.isListenersSendPending("ctrl1"), "a send fenced with the current generation clears the debt")
+	})
+
+	t.Run("forget drops only settled controllers no longer known", func(t *testing.T) {
+		sync := newCtrlSynchronizer()
+
+		// ctrl1: settled, gone from the snapshot -> dropped.
+		sync.markReconnected("ctrl1")
+		gen, _ := sync.beginRefresh("ctrl1")
+		require.True(t, sync.markRefreshWritten("ctrl1", gen))
+		sync.markListenersSent("ctrl1", sync.currentListenersGen(), sync.reconnectGen("ctrl1"))
+
+		// ctrl2: connected after the snapshot was taken, so absent from it, but owing a refresh -> kept.
+		sync.markReconnected("ctrl2")
+
+		// ctrl3: refreshed but still owing the listener set -> kept.
+		sync.markReconnected("ctrl3")
+		gen3, _ := sync.beginRefresh("ctrl3")
+		require.True(t, sync.markRefreshWritten("ctrl3", gen3))
+
+		sync.forget(map[string]struct{}{})
+
+		_, ok1 := sync.states["ctrl1"]
+		require.False(t, ok1, "a settled state for an unknown controller is dropped")
+		require.False(t, sync.isSynced("ctrl2"), "a reconnect recorded after the snapshot keeps its refresh debt")
+		_, owed := sync.beginRefresh("ctrl2")
+		require.True(t, owed)
+		require.True(t, sync.isListenersSendPending("ctrl3"), "an owed listener set is kept")
+	})
+}

router/link/link_events.go

@@ -61,7 +61,9 @@ func (self *removeLinkDest) Handle(registry *linkRegistryImpl) {
 }
 
 type linkDestUpdate struct {
-	id        string
+	id string
+	// ctrlId is the controller the update came from, empty for a synthetic update from a local rescan.
+	ctrlId    string
 	version   string
 	healthy   bool
 	listeners []*ctrl_pb.Listener
@@ -204,6 +206,23 @@ func (self *linkDestUpdate) ApplyListenerChanges(registry *linkRegistryImpl, des
 	for linkKey := range currentLinkKeys {
 		if v, ok := dest.linkMap[linkKey]; ok {
 			delete(dest.linkMap, linkKey)
+
+			// Logged even when nothing is closed: a detached pairing cannot redial or report, so a dial
+			// already in flight lands with nowhere to go, and nothing else records why.
+			source := "local dialer rescan"
+			if closeOrphans {
+				source = "peer listener update"
+			}
+			pfxlog.Logger().
+				WithField("routerId", self.id).
+				WithField("ctrlId", self.ctrlId).
+				WithField("linkKey", linkKey).
+				WithField("linkId", v.linkId).
+				WithField("status", v.status).
+				WithField("hasLink", v.link != nil).
+				WithField("source", source).
+				Info("detaching link pairing that no longer matches")
+
 			if closeOrphans && v.link != nil {
 				log := pfxlog.Logger().WithField("routerId", self.id).
 					WithField("linkKey", linkKey)
@@ -225,7 +244,11 @@ func (self *updateLinkStatusForLink) Handle(registry *linkRegistryImpl) {
 	dest, found := registry.destinations[link.DestinationId()]
 	if !found {
 		if link.IsDialed() { // if link was created by listener, rather than dialer we may not have an entry for it
-			log.WithField("linkDest", link.DestinationId()).Warnf("unable to mark link as %s, link destination not present in registry", self.status)
+			// Reached when a dial completes after its destination has been removed. The link cannot be
+			// recorded, so closing it is the only way it ever ends.
+			log.WithField("linkDest", link.DestinationId()).WithField("status", self.status).
+				Warn("closing dialed link, its destination is not present in registry")
+			registry.closeUnaccountedLink(link)
 		}
 		return
 	}
@@ -241,7 +264,9 @@ func (self *updateLinkStatusForLink) Handle(registry *linkRegistryImpl) {
 				log.WithField("linkDest", link.DestinationId()).Info("reporting fault for closed detached link")
 				registry.sendLinkFaultDirect(link)
 			} else {
-				log.WithField("linkDest", link.DestinationId()).Warnf("unable to mark link as %s, link state not present in registry", self.status)
+				log.WithField("linkDest", link.DestinationId()).WithField("status", self.status).
+					Warn("closing dialed link, its state is not present in registry")
+				registry.closeUnaccountedLink(link)
 			}
 		}
 		return

router/link/link_registry.go

@@ -36,12 +36,16 @@ import (
 	"github.com/openziti/ziti/v2/router/env"
 	"github.com/openziti/ziti/v2/router/xlink"
 	"github.com/sirupsen/logrus"
+	"google.golang.org/protobuf/proto"
 )
 
 type Env interface {
 	GetRouterId() *identity.TokenId
 	GetNetworkControllers() env.NetworkControllers
 	GetXlinkDialers() []xlink.Dialer
+	// GetXlinkListenerSnapshot returns the current link listeners and the config generation they belong to,
+	// read together.
+	GetXlinkListenerSnapshot() (uint64, []xlink.Listener)
 	GetCloseNotify() <-chan struct{}
 	GetLinkDialerPool() goroutines.Pool
 	GetRateLimiterPool() goroutines.Pool
@@ -59,8 +63,8 @@ func NewLinkRegistry(routerEnv Env) xlink.Registry {
 		destinations:           map[string]*linkDest{},
 		linkStateQueue:         &linkStateHeap{},
 		triggerNotifyC:         make(chan struct{}, 1),
+		ctrlSynchronizer:       newCtrlSynchronizer(),
 		fullRefreshSendTimeout: fullRefreshSendTimeout,
-		fullRefreshRetryDelay:  fullRefreshRetryDelay,
 	}
 
 	go result.run()
@@ -78,15 +82,19 @@ type linkRegistryImpl struct {
 
 	env          Env
 	destinations map[string]*linkDest
-	// fullRefreshSendTimeout and fullRefreshRetryDelay bound and pace the reconnect announcement. Fields so
-	// tests can reach the timeout, which is the only way to tell a message that was queued and then discarded
-	// from one that reached the wire.
-	fullRefreshSendTimeout time.Duration
-	fullRefreshRetryDelay  time.Duration
-	linkStateQueue         *linkStateHeap
-	events                 chan event
-	triggerNotifyC         chan struct{}
-	notifyInProgress       atomic.Bool
+	// ctrlSynchronizer holds what each controller is still owed: a full refresh after a reconnect, which gates
+	// incremental reports to it, and the current link listener set.
+	ctrlSynchronizer *ctrlSynchronizer
+
+	// fullRefreshSendTimeout bounds one attempt at the reconnect announcement. A field so tests can reach the
+	// timeout, which is the only way to tell a message that was queued and then discarded from one that
+	// reached the wire.
+	fullRefreshSendTimeout   time.Duration
+	linkStateQueue           *linkStateHeap
+	events                   chan event
+	triggerNotifyC           chan struct{}
+	notifyInProgress         atomic.Bool
+	listenerNotifyInProgress atomic.Bool
 }
 
 func (self *linkRegistryImpl) runGcLinkMetricsLoop() {
@@ -395,54 +403,86 @@ func (self *linkRegistryImpl) Iter() <-chan xlink.Xlink {
 	return result
 }
 
-const (
-	// fullRefreshSendTimeout bounds one attempt at the reconnect announcement. The registry lock is held for
-	// the attempt, so this is also how long link accept and dial-succeeded can stall.
-	fullRefreshSendTimeout = 5 * time.Second
-
-	// fullRefreshSendAttempts bounds the retries. A router announces its full link set once per reconnect and
-	// nothing re-asks, so an announcement that never lands leaves the controller unable to route over those
-	// links and unable to prune the ones it should have dropped.
-	fullRefreshSendAttempts = 3
-
-	fullRefreshRetryDelay = time.Second
-)
+// fullRefreshSendTimeout bounds one attempt at the reconnect announcement. Attempts are not otherwise
+// bounded: a router announces its full link set once per reconnect and nothing re-asks, so an announcement
+// that never lands leaves the controller unable to route over those links and unable to prune the ones it
+// should have dropped. A failed attempt is retried on the registry's next tick.
+const fullRefreshSendTimeout = 5 * time.Second
 
+// NotifyOfReconnect records that a controller has (re)connected and owes a full link announcement. The
+// announcement is sent by the registry loop (see notifyControllersOfRefresh) and retried each tick until it
+// is written. Until then the controller is unsynced and incremental link reports to it are held (see
+// ctrlSynchronizer); once written, the held reports and the current link listeners follow.
 func (self *linkRegistryImpl) NotifyOfReconnect(ch channel.Channel) {
-	// Retried here rather than by clearing the announced marks and letting the periodic path pick it up: that
-	// path sends without FullRefresh, so the controller cannot prune, and it sends nothing at all when there
-	// are no established links, which is exactly when stale controller state most needs clearing.
-	//
-	// Called on its own goroutine (Router.NotifyOfReconnect), so waiting between attempts blocks nothing.
-	for attempt := 1; attempt <= fullRefreshSendAttempts; attempt++ {
-		if self.sendFullRefresh(ch) {
-			return
-		}
-
-		if ch.IsClosed() {
-			return // whatever replaces this channel announces again
-		}
-
-		if attempt < fullRefreshSendAttempts {
-			select {
-			case <-time.After(self.fullRefreshRetryDelay):
-			case <-self.env.GetCloseNotify():
-				return
-			}
-		}
-	}
-
-	pfxlog.Logger().WithField("ctrlId", ch.Id()).
-		Error("gave up announcing link states after reconnect; this controller cannot route over or prune this router's links until it reconnects")
+	self.ctrlSynchronizer.markReconnected(ch.Id())
+	self.triggerNotify()
 }
 
-// sendFullRefresh announces every dialed link to one controller, reporting whether the announcement got out.
-func (self *linkRegistryImpl) sendFullRefresh(ch channel.Channel) bool {
+// notifyControllersOfRefresh queues a full link announcement for every connected controller that owes one
+// and has none in flight. Runs on the registry loop; the sends run on the rate limiter pool.
+func (self *linkRegistryImpl) notifyControllersOfRefresh() {
+	known := map[string]struct{}{}
+	defer func() { self.ctrlSynchronizer.forget(known) }()
+
+	for ctrlId, ctrl := range self.ctrls.GetAll() {
+		known[ctrlId] = struct{}{}
+		if !ctrl.IsConnected() {
+			continue
+		}
+		gen, ok := self.ctrlSynchronizer.beginRefresh(ctrlId)
+		if !ok {
+			continue
+		}
+		ch := ctrl.Channel()
+		err := self.env.GetRateLimiterPool().QueueOrError(func() {
+			self.sendFullRefresh(ctrlId, ch, gen)
+		})
+		if err != nil {
+			pfxlog.Logger().WithError(err).WithField("ctrlId", ctrlId).Info("unable to queue link state announcement")
+			self.ctrlSynchronizer.markRefreshFailed(ctrlId)
+		}
+	}
+}
+
+// sendFullRefresh announces every dialed link to one controller. Each attempt announces the links held at
+// that moment, so a retried announcement is never stale. The link set is snapshotted under the registry
+// lock and sent outside it; ordering against incremental reports is provided by the ctrlSynchronizer gate. gen is
+// the reconnect generation the announcement is for; a reconnect since then leaves the debt in place.
+func (self *linkRegistryImpl) sendFullRefresh(ctrlId string, ch channel.Channel, gen uint64) {
+	log := pfxlog.Logger().WithField("ctrlId", ctrlId)
+	log.Info("resending link states after reconnect")
+
+	routerLinks, onComplete := self.snapshotDialedLinks(ctrlId)
+	log = log.WithField("linkCount", len(routerLinks.Links))
+
+	// SendAndWaitForWire, not Send: Send returns once the message is queued, and the deadline stays live, so
+	// the tx loop discards it if the queue drains too slowly. Nothing is told, since a plain send's listener
+	// ignores the error, and this would report success and mark the links synchronized for an announcement
+	// that never left. Matches how the periodic path sends.
+	if err := protobufs.MarshalTyped(routerLinks).WithTimeout(self.fullRefreshSendTimeout).SendAndWaitForWire(ch); err != nil {
+		log.WithError(err).Warn("link states not yet announced after reconnect, will retry; this controller cannot route over or prune this router's links until they are")
+		self.ctrlSynchronizer.markRefreshFailed(ctrlId)
+		return
+	}
+
+	for _, f := range onComplete {
+		f()
+	}
+
+	if self.ctrlSynchronizer.markRefreshWritten(ctrlId, gen) {
+		log.Info("sent router links on reconnect")
+	} else {
+		log.Info("sent router links on reconnect, but the controller reconnected meanwhile; announcing again")
+	}
+	self.triggerNotify()
+}
+
+// snapshotDialedLinks builds a FullRefresh report of every dialed link, plus the per-link callbacks to run
+// once the report for ctrlId is written.
+func (self *linkRegistryImpl) snapshotDialedLinks(ctrlId string) (*ctrl_pb.RouterLinks, []func()) {
 	self.Lock()
 	defer self.Unlock()
 
-	pfxlog.Logger().WithField("ctrlId", ch.Id()).Info("resending link states after reconnect")
-
 	var onComplete []func()
 
 	// If controllers are just disconnected and not down, they may miss link faults. When they get the complete set
@@ -463,37 +503,92 @@ func (self *linkRegistryImpl) sendFullRefresh(ch channel.Channel) bool {
 
 			if multiConnLink, ok := link.(xlink.MultiConnXLink); ok {
 				onComplete = append(onComplete, func() {
-					multiConnLink.MarkLinkStateSynced(ch.Id())
+					multiConnLink.MarkLinkStateSynced(ctrlId)
 				})
 			}
 		}
 	}
+	return routerLinks, onComplete
+}
 
-	log := logrus.WithField("ctrlId", ch.Id()).WithField("linkCount", len(routerLinks.Links))
+// LinkListenersChanged implements xlink.Registry. The current listener set becomes pending for every
+// controller and is sent by the notify loop, retrying until written.
+func (self *linkRegistryImpl) LinkListenersChanged() {
+	self.ctrlSynchronizer.markListenersChanged()
+	self.triggerNotify()
+}
 
-	// SendAndWaitForWire, not Send: Send returns once the message is queued, and the deadline stays live, so
-	// the tx loop discards it if the queue drains too slowly. Nothing is told, since a plain send's listener
-	// ignores the error, and this would report success and mark the links synchronized for an announcement
-	// that never left. Matches how the periodic path sends.
-	if err := protobufs.MarshalTyped(routerLinks).WithTimeout(self.fullRefreshSendTimeout).SendAndWaitForWire(ch); err != nil {
-		log.WithError(err).Error("failed to send router links on reconnect")
-		return false
+// notifyControllersOfLinkListeners sends the current link listener set to every synced, connected
+// controller that has not been sent it. Runs on the registry loop; the sends run on the rate limiter pool.
+func (self *linkRegistryImpl) notifyControllersOfLinkListeners() {
+	if self.listenerNotifyInProgress.Load() {
+		return
 	}
 
-	log.Info("sent router links on reconnect")
-	for _, f := range onComplete {
-		f()
+	targets := map[string]env.NetworkController{}
+	for ctrlId, ctrl := range self.ctrls.GetAll() {
+		if ctrl.IsConnected() && self.ctrlSynchronizer.isSynced(ctrlId) && self.ctrlSynchronizer.isListenersSendPending(ctrlId) {
+			targets[ctrlId] = ctrl
+		}
+	}
+
+	if len(targets) == 0 {
+		return
+	}
+
+	self.listenerNotifyInProgress.Store(true)
+	err := self.env.GetRateLimiterPool().QueueOrError(func() {
+		defer self.listenerNotifyInProgress.Store(false)
+		self.sendLinkListeners(targets)
+	})
+	if err != nil {
+		pfxlog.Logger().WithError(err).WithField("op", "link-notify").Info("unable to queue link listener notifications")
+		self.listenerNotifyInProgress.Store(false)
+	}
+}
+
+// sendLinkListeners sends the current link listener set to each target, marking the set sent for those it
+// was written to. A set that changes after it is built stays pending, so the next pass sends the newer one.
+func (self *linkRegistryImpl) sendLinkListeners(targets map[string]env.NetworkController) {
+	gen := self.ctrlSynchronizer.currentListenersGen()
+	listeners := ListenersToProto(self.env.GetXlinkListenerSnapshot())
+	buf, err := proto.Marshal(listeners)
+	if err != nil {
+		pfxlog.Logger().WithError(err).WithField("op", "link-notify").Error("unable to marshal link listeners")
+		return
+	}
+
+	for ctrlId, ctrl := range targets {
+		log := pfxlog.Logger().WithField("ctrlId", ctrlId).WithField("op", "link-notify").
+			WithField("listenerCount", len(listeners.Listeners))
+
+		// Captured before the write so a reconnect landing during it leaves the set owed to the new
+		// connection. A controller that went unsynced since it was chosen is skipped; it is sent after its
+		// refresh.
+		reconnectGen := self.ctrlSynchronizer.reconnectGen(ctrlId)
+		if !self.ctrlSynchronizer.isSynced(ctrlId) {
+			continue
+		}
+
+		// A fresh message per controller: sending mutates message state.
+		msg := channel.NewMessage(int32(ctrl_pb.ContentType_UpdateLinkListenersType), buf)
+		if err := msg.WithTimeout(10 * time.Second).SendAndWaitForWire(ctrl.Channel()); err != nil {
+			log.WithError(err).Warn("failed to send link listeners to controller, will retry")
+			continue
+		}
+		self.ctrlSynchronizer.markListenersSent(ctrlId, gen, reconnectGen)
+		log.Info("sent link listeners to controller")
 	}
-	return true
 }
 
 func (self *linkRegistryImpl) GetTraceDecoders() []channel.TraceMessageDecoder {
 	return nil
 }
 
-func (self *linkRegistryImpl) UpdateLinkDest(id string, version string, healthy bool, listeners []*ctrl_pb.Listener) {
+func (self *linkRegistryImpl) UpdateLinkDest(ctrlId string, id string, version string, healthy bool, listeners []*ctrl_pb.Listener) {
 	updateEvent := &linkDestUpdate{
 		id:        id,
+		ctrlId:    ctrlId,
 		version:   version,
 		healthy:   healthy,
 		listeners: listeners,
@@ -570,10 +665,14 @@ func (self *linkRegistryImpl) run() {
 		case evt := <-self.events:
 			evt.Handle(self)
 		case <-self.triggerNotifyC:
+			self.notifyControllersOfRefresh()
 			self.notifyControllersOfLinks()
+			self.notifyControllersOfLinkListeners()
 		case <-queueCheckTicker.C:
 			self.evaluateLinkStateQueue()
+			self.notifyControllersOfRefresh()
 			self.notifyControllersOfLinks()
+			self.notifyControllersOfLinkListeners()
 		case <-fullScanTicker.C:
 			self.evaluateDestinations()
 			self.syncRequiredLinkStates()
@@ -598,6 +697,12 @@ func (self *linkRegistryImpl) evaluateLinkStateQueue() {
 			return
 		}
 		heap.Pop(self.linkStateQueue)
+		// Removing a destination leaves its linkMap intact, so the check below cannot
+		// tell that this state's destination is gone. Dialing anyway reaches whoever
+		// now answers that address.
+		if self.destinations[next.dest.id] != next.dest {
+			continue
+		}
 		// A queued state that is no longer the live entry for its key has been
 		// detached (e.g. by a local dialer rescan). Drop the stale entry rather
 		// than evaluate it, or it would redial through a dialer that rescan has
@@ -609,6 +714,20 @@ func (self *linkRegistryImpl) evaluateLinkStateQueue() {
 	}
 }
 
+// closeUnaccountedLink closes a dialed link the registry has no state to attach to. Such a link is invisible
+// to this router: it is never reported, faulted or closed, while the far end keeps it and uses its id to
+// refuse the dialer's later attempts for the same pair. A link this router accepted is left alone, since the
+// listener side holds no state for one.
+//
+// Closed unconditionally rather than after an IsClosed check. Close is idempotent, and a link that closes
+// between the check and the call would defeat the check anyway.
+func (self *linkRegistryImpl) closeUnaccountedLink(link xlink.Xlink) {
+	if err := link.Close(); err != nil {
+		pfxlog.Logger().WithError(err).WithField("linkId", link.Id()).WithField("linkDest", link.DestinationId()).
+			Error("error closing link with no state in the registry")
+	}
+}
+
 // sendLinkFaultDirect notifies controllers that a dialed link has failed when no
 // linkState remains to carry the fault through the normal notification loop —
 // e.g. a link whose dial state was detached by a local dialer change and then
@@ -898,6 +1017,13 @@ func (self *linkRegistryImpl) sendNewLinks(links []stateAndLink) {
 	for ctrlId, ctrl := range self.ctrls.GetAll() {
 		log := pfxlog.Logger().WithField("ctrlId", ctrlId).WithField("op", "link-notify")
 		if ctrl.IsConnected() {
+			// Held, not skipped: the report stays pending so it follows the refresh once that is written. A
+			// disconnected controller is skipped as before; the refresh it gets on reconnect covers these links.
+			if !self.ctrlSynchronizer.isSynced(ctrlId) {
+				log.Debug("holding new link report until the reconnect announcement is written")
+				allSent = false
+				continue
+			}
 			msgEnv := protobufs.MarshalTyped(routerLinks).WithTimeout(10 * time.Second)
 			if err := msgEnv.SendAndWaitForWire(ctrl.Channel()); err != nil {
 				log.WithError(err).Error("timeout sending new router links")
@@ -944,6 +1070,13 @@ func (self *linkRegistryImpl) sendLinkFaults(list []stateAndFaults) {
 					WithField("iteration", fault.iteration)
 
 				if ctrl.IsConnected() {
+					// Held for a connected controller still owed its refresh. A disconnected one keeps the
+					// brief-outage hold and long-outage expiry below.
+					if !self.ctrlSynchronizer.isSynced(ctrlId) {
+						log.Debug("holding link fault until the reconnect announcement is written")
+						allSent = false
+						continue
+					}
 					msgEnv := protobufs.MarshalTyped(faultMsg).WithTimeout(10 * time.Second)
 					if err := msgEnv.SendAndWaitForWire(ctrl.Channel()); err != nil {
 						log.WithError(err).Error("timeout sending link fault")

router/link/link_registry_test.go

@@ -40,6 +40,7 @@ import (
 	"github.com/openziti/ziti/v2/router/xlink"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
 )
 
 type testEnv struct {
@@ -51,6 +52,8 @@ type testEnv struct {
 
 	dialersMu sync.Mutex
 	dialers   []xlink.Dialer
+
+	listenerGeneration atomic.Uint64
 }
 
 func (self *testEnv) setDialers(d []xlink.Dialer) {
@@ -98,6 +101,10 @@ func (self *testEnv) GetXlinkDialers() []xlink.Dialer {
 	return out
 }
 
+func (self *testEnv) GetXlinkListenerSnapshot() (uint64, []xlink.Listener) {
+	return self.listenerGeneration.Load(), nil
+}
+
 func (self *testEnv) GetCloseNotify() <-chan struct{} {
 	return self.closeNotify
 }
@@ -107,7 +114,7 @@ func (self *testEnv) GetLinkDialerPool() goroutines.Pool {
 }
 
 func (self *testEnv) GetRateLimiterPool() goroutines.Pool {
-	panic("implement me")
+	return self.dialerPool
 }
 
 func (self *testEnv) GetHeartbeatSettings() xlink.HeartbeatSettings {
@@ -499,7 +506,7 @@ func Test_LinkRegistry_RescanForDialOpportunities(t *testing.T) {
 	// matching dialer yet, so the listener doesn't produce a linkState.
 	destId := "peer-router-1"
 	tenv.setDialers(nil)
-	reg.UpdateLinkDest(destId, "v0", true, []*ctrl_pb.Listener{
+	reg.UpdateLinkDest("ctrl1", destId, "v0", true, []*ctrl_pb.Listener{
 		{Address: "tls:peer:6000", Protocol: "tls", Groups: []string{"a"}},
 	})
 
@@ -530,7 +537,7 @@ func Test_LinkRegistry_RescanIsNoopWhenNoMatches(t *testing.T) {
 	// shouldn't create a state because there's no group intersection.
 	destId := "peer-router-2"
 	tenv.setDialers([]xlink.Dialer{&stubDialer{binding: "transport", groups: []string{"b"}}})
-	reg.UpdateLinkDest(destId, "v0", true, []*ctrl_pb.Listener{
+	reg.UpdateLinkDest("ctrl1", destId, "v0", true, []*ctrl_pb.Listener{
 		{Address: "tls:peer:6000", Protocol: "tls", Groups: []string{"a"}},
 	})
 
@@ -552,14 +559,17 @@ func Test_LinkRegistry_RescanIsNoopWhenNoMatches(t *testing.T) {
 // embeds the interface, so any method these tests do not exercise panics rather than returning a zero value.
 type flakySendChannel struct {
 	channel.Channel
-	failures    int
+	failures    atomic.Int32
 	sends       atomic.Int32
 	closed      atomic.Bool
 	closeNotify chan struct{}
+	lastSent    atomic.Pointer[channel.Message]
 }
 
 func newFlakySendChannel(failures int) *flakySendChannel {
-	return &flakySendChannel{failures: failures, closeNotify: make(chan struct{})}
+	result := &flakySendChannel{closeNotify: make(chan struct{})}
+	result.failures.Store(int32(failures))
+	return result
 }
 
 func (self *flakySendChannel) Id() string     { return "ctrl1" }
@@ -569,72 +579,122 @@ func (self *flakySendChannel) IsClosed() bool { return self.closed.Load() }
 func (self *flakySendChannel) CloseNotify() <-chan struct{} { return self.closeNotify }
 
 func (self *flakySendChannel) Send(s channel.Sendable) error {
-	if int(self.sends.Add(1)) <= self.failures {
+	if self.sends.Add(1) <= self.failures.Load() {
 		return errors.New("timeout waiting for space in send queue")
 	}
 	// What the tx loop does once the message is actually written.
+	self.lastSent.Store(s.Msg())
 	s.SendListener().NotifyAfterWrite()
 	return nil
 }
 
 // newReconnectTestRegistry builds a registry with no links and no event loop. An empty link set is a case
 // worth covering rather than avoiding: the reconnect announcement is the only message that prunes, so a
-// router with nothing to report still has to send one to clear stale controller state.
+// router with nothing to report still has to send one to clear stale controller state. Tests drive the
+// loop's passes by hand.
 func newReconnectTestRegistry(t *testing.T) (*linkRegistryImpl, *testEnv) {
 	t.Helper()
 	routerEnv := newTestEnv()
 	t.Cleanup(func() { close(routerEnv.closeNotify) })
-
 	return &linkRegistryImpl{
-		env:            routerEnv,
-		ctrls:          routerEnv.ctrls,
-		destinations:   map[string]*linkDest{},
-		linkMap:        map[string]xlink.Xlink{},
-		triggerNotifyC: make(chan struct{}, 1),
+		env:              routerEnv,
+		ctrls:            routerEnv.ctrls,
+		destinations:     map[string]*linkDest{},
+		linkMap:          map[string]xlink.Xlink{},
+		events:           make(chan event, 16),
+		triggerNotifyC:   make(chan struct{}, 1),
+		ctrlSynchronizer: newCtrlSynchronizer(),
 		// Production's pacing is not this test's concern; the timeout has to stay reachable, though, since it
 		// is what distinguishes a discarded message from a delivered one.
 		fullRefreshSendTimeout: 20 * time.Millisecond,
-		fullRefreshRetryDelay:  0,
 	}, routerEnv
 }
 
+// testCtrls is an env.NetworkControllers double exposing a fixed controller set.
+type testCtrls struct {
+	env.NetworkControllers
+	all map[string]env.NetworkController
+}
+
+func (self *testCtrls) GetAll() map[string]env.NetworkController { return self.all }
+
+// testCtrl is an env.NetworkController double over a test channel.
+type testCtrl struct {
+	env.NetworkController
+	ch           channel.Channel
+	connected    atomic.Bool
+	sinceContact time.Duration
+}
+
+func (self *testCtrl) IsConnected() bool                   { return self.connected.Load() }
+func (self *testCtrl) Channel() channel.Channel            { return self.ch }
+func (self *testCtrl) TimeSinceLastContact() time.Duration { return self.sinceContact }
+
+// withCtrl points the registry at a single connected controller reached over ch.
+func withCtrl(reg *linkRegistryImpl, ctrlId string, ch channel.Channel) *testCtrl {
+	ctrl := &testCtrl{ch: ch}
+	ctrl.connected.Store(true)
+	reg.ctrls = &testCtrls{all: map[string]env.NetworkController{ctrlId: ctrl}}
+	return ctrl
+}
+
+// tickRefresh runs one loop pass of the refresh sender and waits for the attempt it queued, if any, to finish.
+func tickRefresh(t *testing.T, reg *linkRegistryImpl, ctrlId string) {
+	t.Helper()
+	reg.notifyControllersOfRefresh()
+	require.Eventually(t, func() bool { return !reg.ctrlSynchronizer.isRefreshInFlight(ctrlId) }, time.Second, time.Millisecond)
+}
+
+// markRefreshed records a written refresh for ctrlId without sending one.
+func markRefreshed(t *testing.T, reg *linkRegistryImpl, ctrlId string) {
+	t.Helper()
+	gen, ok := reg.ctrlSynchronizer.beginRefresh(ctrlId)
+	require.True(t, ok)
+	require.True(t, reg.ctrlSynchronizer.markRefreshWritten(ctrlId, gen))
+}
+
 // Test_NotifyOfReconnect_RetriesUntilTheAnnouncementLands covers the retry. A router announces its full link
 // set once per controller reconnect and nothing re-asks, so an announcement lost to send-queue back-pressure
 // leaves that controller unable to route over the router's links, and unable to prune the ones it should have
-// dropped, until the next reconnect.
+// dropped, until the next reconnect. Once written, the controller is synced and owes the listener set.
 func Test_NotifyOfReconnect_RetriesUntilTheAnnouncementLands(t *testing.T) {
 	reg, _ := newReconnectTestRegistry(t)
 	ch := newFlakySendChannel(2)
+	withCtrl(reg, "ctrl1", ch)
 
 	reg.NotifyOfReconnect(ch)
+	require.False(t, reg.ctrlSynchronizer.isSynced("ctrl1"))
 
-	require.Equal(t, int32(3), ch.sends.Load(),
-		"the announcement should have been retried until it reached the wire")
+	for i := 0; i < 3; i++ {
+		tickRefresh(t, reg, "ctrl1")
+	}
+
+	require.Equal(t, int32(3), ch.sends.Load(), "the announcement should have been retried until it reached the wire")
+	require.True(t, reg.ctrlSynchronizer.isSynced("ctrl1"))
+	require.True(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"))
+
+	tickRefresh(t, reg, "ctrl1")
+	require.Equal(t, int32(3), ch.sends.Load(), "a written announcement is not sent again")
 }
 
-// Test_NotifyOfReconnect_GivesUpAfterItsAttempts: the retry is bounded, so a channel that never drains does
-// not hold a goroutine indefinitely.
-func Test_NotifyOfReconnect_GivesUpAfterItsAttempts(t *testing.T) {
+// Test_NotifyOfReconnect_KeepsRetryingWhileTheDebtStands: the retry is not bounded by a count. A channel that
+// never drains is retried every pass, with the controller held unsynced, for as long as it is connected.
+func Test_NotifyOfReconnect_KeepsRetryingWhileTheDebtStands(t *testing.T) {
 	reg, _ := newReconnectTestRegistry(t)
 	ch := newFlakySendChannel(math.MaxInt32)
+	ctrl := withCtrl(reg, "ctrl1", ch)
 
 	reg.NotifyOfReconnect(ch)
+	for i := 0; i < 6; i++ {
+		tickRefresh(t, reg, "ctrl1")
+	}
+	require.Equal(t, int32(6), ch.sends.Load(), "the announcement should be retried on every pass")
+	require.False(t, reg.ctrlSynchronizer.isSynced("ctrl1"), "the controller stays unsynced until the announcement lands")
 
-	require.Equal(t, int32(fullRefreshSendAttempts), ch.sends.Load(),
-		"the retry must stop after its attempts rather than looping")
-}
-
-// Test_NotifyOfReconnect_StopsWhenTheChannelCloses: a closed channel cannot be re-announced to, and whatever
-// replaces it announces again, so retrying against it only delays the goroutine's exit.
-func Test_NotifyOfReconnect_StopsWhenTheChannelCloses(t *testing.T) {
-	reg, _ := newReconnectTestRegistry(t)
-	ch := newFlakySendChannel(math.MaxInt32)
-	ch.closed.Store(true)
-
-	reg.NotifyOfReconnect(ch)
-
-	require.Equal(t, int32(1), ch.sends.Load(),
-		"a closed channel should be abandoned after the first failure, not retried")
+	ctrl.connected.Store(false)
+	tickRefresh(t, reg, "ctrl1")
+	require.Equal(t, int32(6), ch.sends.Load(), "a disconnected controller is left owed rather than retried")
+	require.False(t, reg.ctrlSynchronizer.isSynced("ctrl1"))
 }
 
 // acceptThenDiscardChannel models what a real channel does to a message queued behind a backlog: the send is
@@ -650,7 +710,6 @@ func (self *acceptThenDiscardChannel) Id() string                   { return "ct
 func (self *acceptThenDiscardChannel) Label() string                { return "ctrl1" }
 func (self *acceptThenDiscardChannel) IsClosed() bool               { return false }
 func (self *acceptThenDiscardChannel) CloseNotify() <-chan struct{} { return self.closeNotify }
-
 func (self *acceptThenDiscardChannel) Send(channel.Sendable) error {
 	self.sends.Add(1)
 	return nil // queued, and never written
@@ -663,11 +722,232 @@ func (self *acceptThenDiscardChannel) Send(channel.Sendable) error {
 func Test_NotifyOfReconnect_TreatsADiscardedAnnouncementAsFailure(t *testing.T) {
 	reg, _ := newReconnectTestRegistry(t)
 	ch := &acceptThenDiscardChannel{closeNotify: make(chan struct{})}
+	withCtrl(reg, "ctrl1", ch)
 
 	reg.NotifyOfReconnect(ch)
+	for i := 0; i < 3; i++ {
+		tickRefresh(t, reg, "ctrl1")
+	}
 
-	require.Equal(t, int32(fullRefreshSendAttempts), ch.sends.Load(),
+	require.Equal(t, int32(3), ch.sends.Load(),
 		"an announcement accepted into the queue but never written must count as a failure and be retried")
+	require.False(t, reg.ctrlSynchronizer.isSynced("ctrl1"))
+}
+
+// Test_NotifyOfReconnect_OneAnnouncementInFlightPerController: a pass does not queue a second announcement
+// for a controller whose previous one has not finished.
+func Test_NotifyOfReconnect_OneAnnouncementInFlightPerController(t *testing.T) {
+	reg, _ := newReconnectTestRegistry(t)
+	ch := newFlakySendChannel(0)
+	withCtrl(reg, "ctrl1", ch)
+
+	reg.NotifyOfReconnect(ch)
+	_, ok := reg.ctrlSynchronizer.beginRefresh("ctrl1") // an announcement is in flight
+	require.True(t, ok)
+
+	reg.notifyControllersOfRefresh()
+	require.Equal(t, int32(0), ch.sends.Load(), "no second announcement while one is in flight")
+
+	reg.ctrlSynchronizer.markRefreshFailed("ctrl1")
+	tickRefresh(t, reg, "ctrl1")
+	require.Equal(t, int32(1), ch.sends.Load())
+	require.True(t, reg.ctrlSynchronizer.isSynced("ctrl1"))
+}
+
+// Test_NotifyOfReconnect_ReconnectDuringAnnouncementLeavesItOwed: an announcement built before a further
+// reconnect describes a channel state since replaced, so landing it does not sync the controller; the next
+// pass announces again.
+func Test_NotifyOfReconnect_ReconnectDuringAnnouncementLeavesItOwed(t *testing.T) {
+	reg, _ := newReconnectTestRegistry(t)
+	ch := newFlakySendChannel(0)
+	withCtrl(reg, "ctrl1", ch)
+
+	reg.NotifyOfReconnect(ch)
+	gen, ok := reg.ctrlSynchronizer.beginRefresh("ctrl1")
+	require.True(t, ok)
+	reg.NotifyOfReconnect(ch) // reconnect while the announcement for gen is in flight
+
+	reg.sendFullRefresh("ctrl1", ch, gen)
+	require.Equal(t, int32(1), ch.sends.Load())
+	require.False(t, reg.ctrlSynchronizer.isSynced("ctrl1"), "the stale announcement does not satisfy the later reconnect")
+
+	tickRefresh(t, reg, "ctrl1")
+	require.Equal(t, int32(2), ch.sends.Load(), "one more announcement follows the later reconnect")
+	require.True(t, reg.ctrlSynchronizer.isSynced("ctrl1"))
+}
+
+// Test_sendNewLinks_HeldUntilTheReconnectAnnouncementIsWritten covers the gate. A new-link report that reaches
+// a controller ahead of the full refresh is pruned by it, and a fault that arrives ahead of it re-adds a link
+// the refresh omitted, so incremental reports to an unsynced controller stay pending rather than being sent
+// or marked done.
+func Test_sendNewLinks_HeldUntilTheReconnectAnnouncementIsWritten(t *testing.T) {
+	reg, _ := newReconnectTestRegistry(t)
+	ch := newFlakySendChannel(0)
+	withCtrl(reg, "ctrl1", ch)
+	link := &reportTestLink{id: "l1", destId: "peer"}
+	report := []stateAndLink{{state: &linkState{linkId: link.Id()}, link: link}}
+
+	reg.ctrlSynchronizer.markReconnected("ctrl1")
+	reg.sendNewLinks(report)
+	require.Equal(t, int32(0), ch.sends.Load(), "reports to an unsynced controller are held")
+	require.Empty(t, reg.events, "a held report is not marked as notified")
+
+	markRefreshed(t, reg, "ctrl1")
+	reg.sendNewLinks(report)
+	require.Equal(t, int32(1), ch.sends.Load(), "reports flow once the announcement is written")
+	require.Len(t, reg.events, 1, "a delivered report is marked as notified")
+}
+
+// Test_sendLinkFaults_HeldUntilTheReconnectAnnouncementIsWritten is the fault half of the gate.
+func Test_sendLinkFaults_HeldUntilTheReconnectAnnouncementIsWritten(t *testing.T) {
+	reg, _ := newReconnectTestRegistry(t)
+	ch := newFlakySendChannel(0)
+	withCtrl(reg, "ctrl1", ch)
+	faults := []stateAndFaults{{state: &linkState{}, faults: []linkFault{{linkId: "l1", iteration: 1}}}}
+
+	reg.ctrlSynchronizer.markReconnected("ctrl1")
+	reg.sendLinkFaults(faults)
+	require.Equal(t, int32(0), ch.sends.Load())
+	require.Empty(t, reg.events)
+
+	markRefreshed(t, reg, "ctrl1")
+	reg.sendLinkFaults(faults)
+	require.Equal(t, int32(1), ch.sends.Load())
+	require.Len(t, reg.events, 1)
+}
+
+// Test_linkReports_DisconnectedUnsyncedController covers a controller that owes a refresh but is not
+// connected. The sync gate applies only to connected controllers, so it keeps the handling it always had:
+// new-link reports skip it, since the refresh on reconnect covers them, and faults are held through a brief
+// outage and retired for it after a long one. Holding either for the whole outage would re-send them to the
+// healthy controllers on every tick.
+func Test_linkReports_DisconnectedUnsyncedController(t *testing.T) {
+	newReg := func(sinceContact time.Duration) (*linkRegistryImpl, *flakySendChannel) {
+		reg, _ := newReconnectTestRegistry(t)
+		ch := newFlakySendChannel(0)
+		ctrl := withCtrl(reg, "ctrl1", ch)
+		ctrl.connected.Store(false)
+		ctrl.sinceContact = sinceContact
+		reg.ctrlSynchronizer.markReconnected("ctrl1")
+		return reg, ch
+	}
+
+	t.Run("new link reports are retired", func(t *testing.T) {
+		reg, ch := newReg(time.Hour)
+		link := &reportTestLink{id: "l1", destId: "peer"}
+		reg.sendNewLinks([]stateAndLink{{state: &linkState{linkId: link.Id()}, link: link}})
+		require.Equal(t, int32(0), ch.sends.Load())
+		require.Len(t, reg.events, 1, "the report is marked notified")
+	})
+
+	t.Run("faults are held through a brief outage", func(t *testing.T) {
+		reg, ch := newReg(time.Second)
+		reg.sendLinkFaults([]stateAndFaults{{state: &linkState{}, faults: []linkFault{{linkId: "l1", iteration: 1}}}})
+		require.Equal(t, int32(0), ch.sends.Load())
+		require.Empty(t, reg.events)
+	})
+
+	t.Run("faults are retired after a long outage", func(t *testing.T) {
+		reg, ch := newReg(time.Hour)
+		reg.sendLinkFaults([]stateAndFaults{{state: &linkState{}, faults: []linkFault{{linkId: "l1", iteration: 1}}}})
+		require.Equal(t, int32(0), ch.sends.Load())
+		require.Len(t, reg.events, 1, "the fault is marked notified")
+	})
+}
+
+// Test_sendLinkListeners covers the listener publish: a controller owes the set after a reconnect and after a
+// change, a written set clears what it owes, a failed write leaves it owed, and a change made while a send is
+// being built is not cleared by that send. The loop pass only sends to synced controllers.
+func Test_sendLinkListeners(t *testing.T) {
+	t.Run("a reconnect leaves the set owed until the refresh is written and the set sent", func(t *testing.T) {
+		reg, _ := newReconnectTestRegistry(t)
+		ch := newFlakySendChannel(0)
+		withCtrl(reg, "ctrl1", ch)
+
+		reg.ctrlSynchronizer.markReconnected("ctrl1")
+		reg.notifyControllersOfLinkListeners()
+		require.Eventually(t, func() bool { return !reg.listenerNotifyInProgress.Load() }, time.Second, time.Millisecond)
+		require.Equal(t, int32(0), ch.sends.Load(), "listeners wait for the refresh")
+
+		markRefreshed(t, reg, "ctrl1")
+		reg.notifyControllersOfLinkListeners()
+		require.Eventually(t, func() bool { return !reg.ctrlSynchronizer.isListenersSendPending("ctrl1") }, time.Second, time.Millisecond)
+		require.Equal(t, int32(1), ch.sends.Load())
+	})
+
+	t.Run("the set is stamped with the config generation it was read with", func(t *testing.T) {
+		reg, routerEnv := newReconnectTestRegistry(t)
+		routerEnv.listenerGeneration.Store(7)
+		ch := newFlakySendChannel(0)
+		ctrl := withCtrl(reg, "ctrl1", ch)
+		reg.ctrlSynchronizer.markReconnected("ctrl1")
+		markRefreshed(t, reg, "ctrl1")
+
+		reg.sendLinkListeners(map[string]env.NetworkController{"ctrl1": ctrl})
+
+		msg := ch.lastSent.Load()
+		require.NotNil(t, msg)
+		sent := &ctrl_pb.Listeners{}
+		require.NoError(t, proto.Unmarshal(msg.Body, sent))
+		require.Equal(t, uint64(7), sent.Generation)
+	})
+
+	t.Run("a change makes the set owed again and a failed write leaves it owed", func(t *testing.T) {
+		reg, _ := newReconnectTestRegistry(t)
+		ch := newFlakySendChannel(1)
+		ctrl := withCtrl(reg, "ctrl1", ch)
+		reg.ctrlSynchronizer.markReconnected("ctrl1")
+		markRefreshed(t, reg, "ctrl1")
+		reg.ctrlSynchronizer.markListenersSent("ctrl1", reg.ctrlSynchronizer.currentListenersGen(), reg.ctrlSynchronizer.reconnectGen("ctrl1"))
+		require.False(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"))
+
+		reg.LinkListenersChanged()
+		require.True(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"))
+
+		reg.sendLinkListeners(map[string]env.NetworkController{"ctrl1": ctrl})
+		require.True(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"), "a failed write leaves the set owed")
+
+		reg.sendLinkListeners(map[string]env.NetworkController{"ctrl1": ctrl})
+		require.False(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"))
+	})
+
+	t.Run("a controller that reconnected since it was chosen is not sent to", func(t *testing.T) {
+		reg, _ := newReconnectTestRegistry(t)
+		ch := newFlakySendChannel(0)
+		ctrl := withCtrl(reg, "ctrl1", ch)
+		reg.ctrlSynchronizer.markReconnected("ctrl1")
+		markRefreshed(t, reg, "ctrl1")
+
+		reg.ctrlSynchronizer.markReconnected("ctrl1") // reconnect after the target was chosen
+		reg.sendLinkListeners(map[string]env.NetworkController{"ctrl1": ctrl})
+
+		require.Equal(t, int32(0), ch.sends.Load(), "an unsynced controller waits for its refresh")
+		require.True(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"))
+	})
+
+	t.Run("a reconnect during the write leaves the new connection owed", func(t *testing.T) {
+		reg, _ := newReconnectTestRegistry(t)
+		ch := &reconnectOnSendChannel{flakySendChannel: newFlakySendChannel(0), reg: reg}
+		ctrl := withCtrl(reg, "ctrl1", ch)
+		reg.ctrlSynchronizer.markReconnected("ctrl1")
+		markRefreshed(t, reg, "ctrl1")
+
+		reg.sendLinkListeners(map[string]env.NetworkController{"ctrl1": ctrl})
+
+		require.Equal(t, int32(1), ch.sends.Load())
+		require.True(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"), "the write reached the previous connection")
+	})
+
+	t.Run("a change during a send is not cleared by that send", func(t *testing.T) {
+		reg, _ := newReconnectTestRegistry(t)
+		gen := reg.ctrlSynchronizer.currentListenersGen()
+		reg.ctrlSynchronizer.markReconnected("ctrl1")
+
+		reg.LinkListenersChanged()
+		reg.ctrlSynchronizer.markListenersSent("ctrl1", gen, reg.ctrlSynchronizer.reconnectGen("ctrl1"))
+
+		require.True(t, reg.ctrlSynchronizer.isListenersSendPending("ctrl1"), "the send built before the change cannot satisfy it")
+	})
 }
 
 func Test_LinkRegistry_RescanDetachesRemovedDialerPairing(t *testing.T) {
@@ -680,7 +960,7 @@ func Test_LinkRegistry_RescanDetachesRemovedDialerPairing(t *testing.T) {
 	// A matching dialer + listener produces a linkState.
 	destId := "peer-router-3"
 	tenv.setDialers([]xlink.Dialer{&stubDialer{binding: "transport", groups: []string{"a"}}})
-	reg.UpdateLinkDest(destId, "v0", true, []*ctrl_pb.Listener{
+	reg.UpdateLinkDest("ctrl1", destId, "v0", true, []*ctrl_pb.Listener{
 		{Address: "tls:peer:6000", Protocol: "tls", Groups: []string{"a"}},
 	})
 	req.Eventually(func() bool {
@@ -815,7 +1095,7 @@ func Test_LinkRegistry_GetDestinationListeners_OmitsUnhealthyDestinations(t *tes
 	reg := NewLinkRegistry(tenv).(*linkRegistryImpl)
 
 	destId := "peer-router-1"
-	reg.UpdateLinkDest(destId, "v0", true, []*ctrl_pb.Listener{
+	reg.UpdateLinkDest("ctrl1", destId, "v0", true, []*ctrl_pb.Listener{
 		{Address: "tls:peer:6000", Protocol: "tls", Groups: []string{"a"}},
 	})
 
@@ -826,7 +1106,7 @@ func Test_LinkRegistry_GetDestinationListeners_OmitsUnhealthyDestinations(t *tes
 
 	// PeerState_Unhealthy carries no listeners, so the cache empties. It has to
 	// drop out entirely rather than appear with an empty set.
-	reg.UpdateLinkDest(destId, "v0", false, nil)
+	reg.UpdateLinkDest("ctrl1", destId, "v0", false, nil)
 
 	req.Eventually(func() bool {
 		snapshot, ok := reg.GetDestinationListeners()
@@ -858,3 +1138,111 @@ func Test_LinkRegistry_ReconcilesHeartbeatsOnRegistration(t *testing.T) {
 	req.Len(pushes, 1, "registration must push the current generation")
 	req.Equal(tenv.GetHeartbeatSettings(), pushes[0])
 }
+
+// reportTestLink is an xlink.Xlink double carrying only what a new-link report reads.
+type reportTestLink struct {
+	xlink.Xlink
+	id     string
+	destId string
+}
+
+func (self *reportTestLink) Id() string                               { return self.id }
+func (self *reportTestLink) DestinationId() string                    { return self.destId }
+func (self *reportTestLink) LinkProtocol() string                     { return "tls" }
+func (self *reportTestLink) DialAddress() string                      { return "tls:peer:6000" }
+func (self *reportTestLink) Iteration() uint32                        { return 1 }
+func (self *reportTestLink) IsDialed() bool                           { return true }
+func (self *reportTestLink) GetLinkConnState() *ctrl_pb.LinkConnState { return nil }
+
+// reconnectOnSendChannel records a reconnect of ctrl1 as each write completes, modeling a reconnect that lands
+// between a write and its completion being recorded.
+type reconnectOnSendChannel struct {
+	*flakySendChannel
+	reg *linkRegistryImpl
+}
+
+func (self *reconnectOnSendChannel) Send(s channel.Sendable) error {
+	self.reg.ctrlSynchronizer.markReconnected("ctrl1")
+	return self.flakySendChannel.Send(s)
+}
+
+// Test_LinkRegistry_UnaccountedDialedLinkIsClosed: a dial that completes after its destination has been
+// removed cannot be recorded. Only warning left the link up and unreported, while the far end kept it and
+// used its id to refuse the dialer's later attempts for the same pair.
+func Test_LinkRegistry_UnaccountedDialedLinkIsClosed(t *testing.T) {
+	req := require.New(t)
+	tenv := newTestEnv()
+	defer close(tenv.closeNotify)
+
+	reg := NewLinkRegistry(tenv).(*linkRegistryImpl)
+
+	link := &stubXlink{id: "orphan", dialed: true, destId: "removed-dest"}
+	(&updateLinkStatusForLink{link: link, status: StatusEstablished}).Handle(reg)
+
+	req.True(link.IsClosed(), "a dialed link with no destination to belong to must be closed, not just logged")
+}
+
+// Test_LinkRegistry_UnaccountedAcceptedLinkIsLeftOpen: the listener side keeps no state for a link it
+// accepted, so absence there is normal and says nothing about whether the link is wanted.
+func Test_LinkRegistry_UnaccountedAcceptedLinkIsLeftOpen(t *testing.T) {
+	req := require.New(t)
+	tenv := newTestEnv()
+	defer close(tenv.closeNotify)
+
+	reg := NewLinkRegistry(tenv).(*linkRegistryImpl)
+
+	link := &stubXlink{id: "accepted", dialed: false, destId: "no-state-here"}
+	(&updateLinkStatusForLink{link: link, status: StatusEstablished}).Handle(reg)
+
+	req.False(link.IsClosed(), "an accepted link must not be closed for having no dial state")
+}
+
+// Test_LinkRegistry_RemovedDestIsNotDialed: removing a destination leaves its linkMap intact, so a state
+// still sitting in the dial queue looked live and got dialed anyway. That dial reaches whoever now answers
+// the address, and its result has no state to be recorded against, which is how the orphan above is made.
+func Test_LinkRegistry_RemovedDestIsNotDialed(t *testing.T) {
+	req := require.New(t)
+	tenv := newTestEnv()
+	defer close(tenv.closeNotify)
+
+	tenv.setDialers([]xlink.Dialer{&stubDialer{binding: "transport", groups: []string{"default"}}})
+	reg := NewLinkRegistry(tenv).(*linkRegistryImpl)
+	destId := "doomed-dest"
+
+	reg.UpdateLinkDest("ctrl1", destId, "v0", true, []*ctrl_pb.Listener{
+		{Address: "tls:peer:6000", Protocol: "tls", Groups: []string{"default"}, LocalBinding: "default"},
+	})
+	req.Eventually(func() bool { return destLinkCount(reg, destId) == 1 }, 2*time.Second, 25*time.Millisecond)
+
+	stateC := make(chan *linkState, 1)
+	reg.queueEvent(funcEvent(func(r *linkRegistryImpl) {
+		for _, state := range r.destinations[destId].linkMap {
+			stateC <- state
+			return
+		}
+		stateC <- nil
+	}))
+	state := <-stateC
+	req.NotNil(state)
+
+	reg.RemoveLinkDest(destId)
+
+	// Queue it as the only due entry: scheduled to dial, with its destination removed out from under it.
+	statusC := make(chan linkStatus, 1)
+	reg.queueEvent(funcEvent(func(r *linkRegistryImpl) {
+		state.dialActive.Store(false)
+		state.nextDial = time.Now().Add(-time.Minute)
+		*r.linkStateQueue = linkStateHeap{}
+		heap.Push(r.linkStateQueue, state)
+		r.evaluateLinkStateQueue()
+		statusC <- state.status
+	}))
+
+	req.Equal(StatusDestRemoved, <-statusC, "a removed destination's link state must be left alone, not dialed")
+}
+
+// funcEvent runs an arbitrary function on the registry's event loop, so a test can read or poke registry
+// state without racing the loop.
+type funcEvent func(*linkRegistryImpl)
+
+func (self funcEvent) Handle(registry *linkRegistryImpl) { self(registry) }

router/link/stale_gc_test.go

@@ -53,19 +53,20 @@ func (r *gcStubRegistry) GetDestinationListeners() (map[string][]*ctrl_pb.Listen
 }
 
 // The rest of xlink.Registry isn't exercised by RunStaleLinkGc.
-func (*gcStubRegistry) UpdateLinkDest(string, string, bool, []*ctrl_pb.Listener) {}
-func (*gcStubRegistry) RemoveLinkDest(string)                                    {}
-func (*gcStubRegistry) GetLink(string) (xlink.Xlink, bool)                       { return nil, false }
-func (*gcStubRegistry) GetLinkById(string) (xlink.Xlink, bool)                   { return nil, false }
-func (*gcStubRegistry) DialSucceeded(xlink.Xlink) (xlink.Xlink, bool)            { return nil, false }
-func (*gcStubRegistry) LinkAccepted(xlink.Xlink) (xlink.Xlink, bool)             { return nil, false }
-func (*gcStubRegistry) LinkClosed(xlink.Xlink)                                   {}
-func (*gcStubRegistry) Shutdown()                                                {}
-func (*gcStubRegistry) SendRouterLinkMessage(xlink.Xlink, ...channel.Channel)    {}
-func (*gcStubRegistry) Inspect(time.Duration) *inspect.LinksInspectResult        { return nil }
-func (*gcStubRegistry) DebugForgetLink(string) bool                              { return false }
-func (*gcStubRegistry) GetLinkKey(string, string, string, string) string         { return "" }
-func (*gcStubRegistry) RescanForDialOpportunities()                              {}
+func (*gcStubRegistry) UpdateLinkDest(string, string, string, bool, []*ctrl_pb.Listener) {}
+func (*gcStubRegistry) RemoveLinkDest(string)                                            {}
+func (*gcStubRegistry) LinkListenersChanged()                                            {}
+func (*gcStubRegistry) GetLink(string) (xlink.Xlink, bool)                               { return nil, false }
+func (*gcStubRegistry) GetLinkById(string) (xlink.Xlink, bool)                           { return nil, false }
+func (*gcStubRegistry) DialSucceeded(xlink.Xlink) (xlink.Xlink, bool)                    { return nil, false }
+func (*gcStubRegistry) LinkAccepted(xlink.Xlink) (xlink.Xlink, bool)                     { return nil, false }
+func (*gcStubRegistry) LinkClosed(xlink.Xlink)                                           {}
+func (*gcStubRegistry) Shutdown()                                                        {}
+func (*gcStubRegistry) SendRouterLinkMessage(xlink.Xlink, ...channel.Channel)            {}
+func (*gcStubRegistry) Inspect(time.Duration) *inspect.LinksInspectResult                { return nil }
+func (*gcStubRegistry) DebugForgetLink(string) bool                                      { return false }
+func (*gcStubRegistry) GetLinkKey(string, string, string, string) string                 { return "" }
+func (*gcStubRegistry) RescanForDialOpportunities()                                      {}
 
 // gcStubEnv satisfies link.XlinkEnv.
 type gcStubEnv struct {

router/link/staleness_test.go

@@ -39,6 +39,7 @@ type stubXlink struct {
 	linkProtocol string
 	linkKey      xlink.LinkKey
 	closed       bool
+	destId       string
 }
 
 func (*stubXlink) GetDestinationType() string                                            { return "link" }
@@ -57,7 +58,7 @@ func (s *stubXlink) IsClosed() bool         { return s.closed }
 
 // Required by the interface but not exercised by these tests.
 func (*stubXlink) Iteration() uint32           { return 0 }
-func (*stubXlink) DestinationId() string       { return "" }
+func (s *stubXlink) DestinationId() string     { return s.destId }
 func (*stubXlink) DestVersion() string         { return "" }
 func (*stubXlink) CloseOnce(func())            {}
 func (s *stubXlink) Close() error              { s.closed = true; return nil }

router/router.go

@@ -166,6 +166,12 @@ func (self *Router) GetXlinkListeners() []xlink.Listener {
 	return self.linkSubsystem.Listeners()
 }
 
+// GetXlinkListenerSnapshot returns the current link listeners and the config generation they belong to, read
+// under one lock.
+func (self *Router) GetXlinkListenerSnapshot() (uint64, []xlink.Listener) {
+	return self.linkSubsystem.ListenerSnapshot()
+}
+
 func (self *Router) GetXrctrls() []env.Xrctrl {
 	return self.xrctrls
 }
@@ -861,33 +867,17 @@ func (self *Router) registerPlugins() error {
 	return nil
 }
 
-// currentLinkListeners builds a ctrl_pb.Listeners snapshot from the
-// linkSubsystem's current listener slice. Shared between Hello-time
-// header packing and the post-Apply publish path so both wire shapes
-// stay aligned.
+// currentLinkListeners builds the hello ListenersHeader snapshot from the link subsystem's current listeners.
 func (self *Router) currentLinkListeners() *ctrl_pb.Listeners {
-	// Generation and listeners come from one read: an update carries the
-	// generation of the set it actually contains, so a controller comparing
-	// generations can drop a superseded one.
-	generation, current := self.linkSubsystem.ListenerSnapshot()
-	listeners := &ctrl_pb.Listeners{Generation: generation}
-	for _, listener := range current {
-		listeners.Listeners = append(listeners.Listeners, &ctrl_pb.Listener{
-			Address:      listener.GetAdvertisement(),
-			Protocol:     listener.GetLinkProtocol(),
-			Groups:       listener.GetGroups(),
-			LocalBinding: listener.GetLocalBinding(),
-		})
-	}
-	return listeners
+	return link.ListenersToProto(self.GetXlinkListenerSnapshot())
 }
 
 // onLinkSubsystemChanged is invoked after every Apply/Remove that
-// actually mutates the link subsystem. Listener changes get republished
-// to every connected controller so peer routers see the new state via
-// PeerStateChange. Dialer changes trigger a local rescan against known
-// peers, in case the new dialer set unlocks previously-unmatched
-// listeners. Under a non-preserve gcMode the change also drives an auto-GC
+// actually mutates the link subsystem. Listener changes are handed to the
+// link registry, which republishes them to every controller so peer
+// routers see the new state via PeerStateChange. Dialer changes trigger a
+// local rescan against known peers, in case the new dialer set unlocks
+// previously-unmatched listeners. Under a non-preserve gcMode the change also drives an auto-GC
 // pass over the xlink registry, closing entries it just made stale.
 //
 // The mode comes from the change rather than from the active config, so a
@@ -896,7 +886,7 @@ func (self *Router) currentLinkListeners() *ctrl_pb.Listeners {
 // unsupportable, so under orphaned or changed they are all closed.
 func (self *Router) onLinkSubsystemChanged(change link.ConfigurationChange) {
 	if change.ListenersChanged {
-		self.publishLinkListeners()
+		self.xlinkRegistry.LinkListenersChanged()
 	}
 	if change.DialersChanged {
 		self.xlinkRegistry.RescanForDialOpportunities()
@@ -913,29 +903,6 @@ func (self *Router) onLinkSubsystemChanged(change link.ConfigurationChange) {
 	link.RunStaleLinkGc(self, change.GcMode)
 }
 
-// publishLinkListeners marshals the current listener set and sends an
-// UpdateLinkListeners message to every connected controller. The
-// controller's handler updates Router.Listeners + triggers peer
-// redistribution. Hello continues to carry the initial snapshot at
-// connect time; this message handles mid-session changes.
-func (self *Router) publishLinkListeners() {
-	listeners := self.currentLinkListeners()
-	buf, err := proto.Marshal(listeners)
-	if err != nil {
-		pfxlog.Logger().WithError(err).Error("unable to marshal Listeners for UpdateLinkListeners")
-		return
-	}
-	self.ctrls.ForEach(func(ctrlId string, ch channel.Channel) {
-		// Build a fresh message per controller: sending mutates message state,
-		// so a single shared message can't be sent to multiple channels safely.
-		msg := channel.NewMessage(int32(ctrl_pb.ContentType_UpdateLinkListenersType), buf)
-		if err := ch.Send(msg); err != nil {
-			pfxlog.Logger().WithError(err).WithField("ctrlId", ctrlId).
-				Warn("failed to send UpdateLinkListeners to controller")
-		}
-	})
-}
-
 // applyLocalLinkConfig translates the router's local YAML link config into
 // the router.link.v1 JSON shape and feeds it to the managedconfig registry
 // as a local-source apply. Returns immediately if no local link config is

router/xlink/xlink.go

@@ -30,8 +30,8 @@ import (
 
 // Registry contains known link instances and manages link de-duplication
 type Registry interface {
-	// UpdateLinkDest adds or updates the state of the given destination
-	UpdateLinkDest(id string, version string, healthy bool, listeners []*ctrl_pb.Listener)
+	// UpdateLinkDest adds or updates the state of the given destination, as reported by controller ctrlId
+	UpdateLinkDest(ctrlId string, id string, version string, healthy bool, listeners []*ctrl_pb.Listener)
 
 	// RemoveLinkDest removes the given link destination
 	RemoveLinkDest(id string)
@@ -54,6 +54,10 @@ type Registry interface {
 	// SendRouterLinkMessage will notify the given controllers about the existing link
 	SendRouterLinkMessage(link Xlink, channels ...channel.Channel)
 
+	// LinkListenersChanged notifies the registry that the router's link listener set changed, so it is
+	// republished to every controller
+	LinkListenersChanged()
+
 	// Inspect will return debug information about the state of links and the registry
 	Inspect(timeout time.Duration) *inspect.LinksInspectResult
 

router/xlink_transport/xlink_split.go

@@ -185,8 +185,17 @@ func (self *splitImpl) CloseOnce(f func()) {
 	}
 }
 
+// IsClosed reports whether either channel of the link has closed. A channel that is not yet bound counts as
+// open: the listener side sets the two independently as each binds, so a link is briefly half-built, and
+// Close skips a nil channel for the same reason.
 func (self *splitImpl) IsClosed() bool {
-	return self.payloadCh.IsClosed() || self.ackCh.IsClosed()
+	if ch := self.payloadCh; ch != nil && ch.IsClosed() {
+		return true
+	}
+	if ch := self.ackCh; ch != nil && ch.IsClosed() {
+		return true
+	}
+	return false
 }
 
 func (self *splitImpl) IsDialed() bool {

router/xlink_transport/xlink_split_test.go

@@ -0,0 +1,33 @@
+/*
+	Copyright NetFoundry Inc.
+
+	Licensed under the Apache License, Version 2.0 (the "License");
+	you may not use this file except in compliance with the License.
+	You may obtain a copy of the License at
+
+	https://www.apache.org/licenses/LICENSE-2.0
+
+	Unless required by applicable law or agreed to in writing, software
+	distributed under the License is distributed on an "AS IS" BASIS,
+	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+	See the License for the specific language governing permissions and
+	limitations under the License.
+*/
+
+package xlink_transport
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestSplitImpl_IsClosedOnHalfBuiltLink: the listener binds a split link's two channels independently, as
+// each arrives, so the link exists with only one of them set. Close has always skipped a nil channel, while
+// IsClosed dereferenced both, which made asking whether such a link was closed a panic.
+func TestSplitImpl_IsClosedOnHalfBuiltLink(t *testing.T) {
+	req := require.New(t)
+
+	req.NotPanics(func() { _ = (&splitImpl{}).IsClosed() })
+	req.False((&splitImpl{}).IsClosed(), "a link with neither channel bound is not closed")
+}

tests/link_test.go

@@ -71,6 +71,10 @@ func (self *testRegistryEnv) GetXlinkDialers() []xlink.Dialer {
 	panic("implement me")
 }
 
+func (self *testRegistryEnv) GetXlinkListenerSnapshot() (uint64, []xlink.Listener) {
+	return 0, nil
+}
+
 func (self *testRegistryEnv) GetCloseNotify() <-chan struct{} {
 	return self.closeNotify
 }

tests/testutil/linkschecker.go

@@ -107,9 +107,10 @@ func (self *LinkStateChecker) HandleOther(msg *channel.Message, _ channel.Channe
 	//    5 = heartbeat
 	// 1007 = metrics message
 	// 1053 = LinkState
+	// 1056 = UpdateLinkListeners
 	// 201415 = connect events
 	if msg.ContentType == -33 || msg.ContentType == 5 || msg.ContentType == 1007 || msg.ContentType == 1053 ||
-		msg.ContentType == 20415 {
+		msg.ContentType == 1056 || msg.ContentType == 20415 {
 		logrus.Debug("ignoring heartbeats, reconnect pings and metrics")
 		return
 	}

zititest/models/circuit-perf-diff/configs/router.yml.tmpl

@@ -39,7 +39,7 @@ link:
     - binding:          transport
       split: false
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/circuit-test/configs/router.yml.tmpl

@@ -42,7 +42,7 @@ link:
     - binding:          transport
       split: false
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/circuit-test/main.go

@@ -145,7 +145,7 @@ var m = &model.Model{
 	Id: "circuit-test",
 	Scope: model.Scope{
 		Defaults: model.Variables{
-			"ha":          "false",
+			"ha":          "true",
 			"tcpdump":     "false",
 			"environment": "circuit-test",
 			"credentials": model.Variables{

zititest/models/circuit-test/sim_metrics.go

@@ -67,6 +67,8 @@ func (self *SimMetricsValidator) AcceptHostMetrics(host *model.Host, event *mode
 				metrics.PopulateTimer(&metrics.latency, set)
 			case "tx.bytes":
 				metrics.PopulateMeter(&metrics.throughput, set)
+			case "tx.byterate":
+				metrics.PopulateGaugeFloat64(&metrics.byteRate, set)
 			case "tx.messages", "rx.bytes", "rx.messages", "active", "completed":
 			default:
 				fmt.Printf("ignoring metric %s for service %s\n", metricName, serviceName)
@@ -99,6 +101,11 @@ func (self *SimMetricsValidator) AcceptHostMetrics(host *model.Host, event *mode
 			metric.throughput = &model.Meter{}
 			errList = append(errList, fmt.Errorf("missing throughput metric for service %s", service))
 		}
+		if metric.byteRate == nil {
+			zero := float64(0)
+			metric.byteRate = &zero
+			errList = append(errList, fmt.Errorf("missing byterate metric for service %s", service))
+		}
 	}
 	if len(errList) > 0 {
 		for name := range event.Metrics {
@@ -179,6 +186,7 @@ func (self *SimMetricsValidator) ValidateCollected() error {
 	var errList []error
 
 	for host, events := range self.events {
+		byteRates := lastByteRates(events)
 		for idx, event := range events {
 			isLast := idx == len(events)-1
 			for service, metrics := range event.Metrics {
@@ -233,12 +241,18 @@ func (self *SimMetricsValidator) ValidateCollected() error {
 					}
 				}
 
+				// Throughput is judged on the whole-transfer rate the sim reports when a workload finishes, not
+				// on the tx.bytes meter's one-minute EWMA: a throughput workload completes in well under a
+				// minute, so the EWMA never converges and its reading depends on where the burst falls
+				// against the sampling ticks.
 				if isLast && strings.Contains(service, "throughput") {
 					throughputThreshold := throughputThresholds[clientType]
-
-					if uint64(metrics.throughput.M1Rate) < throughputThreshold {
+					byteRate, found := byteRates[service]
+					if !found {
+						errList = append(errList, fmt.Errorf("%s: service %s reported no transfer rate", host.Id, service))
+					} else if uint64(byteRate) < throughputThreshold {
 						errList = append(errList, fmt.Errorf("%s: service %s has throughput %v not meeting %s",
-							host.Id, service, outputz.FormatBytes(uint64(metrics.throughput.M1Rate)),
+							host.Id, service, outputz.FormatBytes(uint64(byteRate)),
 							outputz.FormatBytes(throughputThreshold)))
 					}
 				}
@@ -248,7 +262,8 @@ func (self *SimMetricsValidator) ValidateCollected() error {
 						fmt.Printf("%s: service %s has mean latency %s, p95 latency %s\n", host.Id, service,
 							time.Duration(int64(metrics.latency.Mean)).String(), time.Duration(int64(metrics.latency.P95)).String())
 					} else {
-						fmt.Printf("%s: service %s has throughput %v\n", host.Id, service,
+						fmt.Printf("%s: service %s has throughput %v (1m ewma %v)\n", host.Id, service,
+							outputz.FormatBytes(uint64(byteRates[service])),
 							outputz.FormatBytes(uint64(metrics.throughput.M1Rate)))
 					}
 				}
@@ -295,6 +310,8 @@ func (self *SimMetricsValidator) LogMetrics() {
 				}
 				if !strings.Contains(service, "latency") {
 					self.LogMeter("tx.bytes", metrics.throughput)
+					self.printIndentF(8, "gauge: tx.byterate")
+					self.printIndentF(12, "value    : %f", *metrics.byteRate)
 				}
 			}
 		}
@@ -356,9 +373,37 @@ type ServiceMetrics struct {
 	connectTimes *model.Timer
 	latency      *model.Timer
 	throughput   *model.Meter
+	byteRate     *float64
 	err          error
 }
 
+// lastByteRates returns, per service, the last non-zero tx.byterate across events. The sim sets the gauge
+// once, when a workload finishes, so earlier events carry zero.
+func lastByteRates(events []*MetricsEvent) map[string]float64 {
+	result := map[string]float64{}
+	for _, event := range events {
+		for service, metrics := range event.Metrics {
+			if metrics.byteRate != nil && *metrics.byteRate > 0 {
+				result[service] = *metrics.byteRate
+			}
+		}
+	}
+	return result
+}
+
+// PopulateGaugeFloat64 reads a float gauge, which the metrics reporter sends as a single "value" entry.
+func (self *ServiceMetrics) PopulateGaugeFloat64(val **float64, metric model.MetricSet) {
+	switch v := metric["value"].(type) {
+	case float64:
+		*val = &v
+	case int64:
+		f := float64(v)
+		*val = &f
+	default:
+		self.err = fmt.Errorf("gauge has no numeric value (%T)", metric["value"])
+	}
+}
+
 func (self *ServiceMetrics) PopulateMeter(val **model.Meter, metric model.MetricSet) {
 	meter, err := metric.AsMeter()
 	if err != nil {

zititest/models/circuit-test/sim_metrics_test.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/openziti/fablab/kernel/model"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_PopulateGaugeFloat64(t *testing.T) {
+	req := require.New(t)
+
+	metrics := &ServiceMetrics{}
+	metrics.PopulateGaugeFloat64(&metrics.byteRate, model.MetricSet{"value": 1234.5})
+	req.NoError(metrics.err)
+	req.Equal(1234.5, *metrics.byteRate)
+
+	metrics = &ServiceMetrics{}
+	metrics.PopulateGaugeFloat64(&metrics.byteRate, model.MetricSet{})
+	req.Error(metrics.err, "a gauge with no value is reported")
+	req.Nil(metrics.byteRate)
+}
+
+// Test_lastByterates: the sim sets the byterate gauge once, when a workload finishes, so earlier events carry
+// zero and the reading that counts is the last non-zero one.
+func Test_lastByterates(t *testing.T) {
+	req := require.New(t)
+	rate := func(v float64) *ServiceMetrics { return &ServiceMetrics{byteRate: &v} }
+
+	events := []*MetricsEvent{
+		{Metrics: map[string]*ServiceMetrics{"throughput-xg": rate(0), "latency-xg": rate(0)}},
+		{Metrics: map[string]*ServiceMetrics{"throughput-xg": rate(55e6), "latency-xg": rate(0)}},
+		{Metrics: map[string]*ServiceMetrics{"throughput-xg": rate(0)}},
+	}
+
+	result := lastByteRates(events)
+	req.Equal(55e6, result["throughput-xg"])
+	_, found := result["latency-xg"]
+	req.False(found, "a service whose gauge was never set reports no rate")
+}

zititest/models/db-ctrl-test/configs/router.yml.tmpl

@@ -43,7 +43,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/ert-hosting-test/configs/router.yml.tmpl

@@ -47,7 +47,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
       groups:
         - {{ .Host.Id }}
         - public

zititest/models/links-test/configs/router.yml.tmpl

@@ -44,7 +44,7 @@ link:
       healthyDialBackoff:
         maxRetryInterval: 1m
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/links-test/main.go

@@ -114,7 +114,7 @@ var m = &model.Model{
 					c.Type = &clone
 					// fmt.Printf("%s: %d - %s - \n", c.Id, c.ScaleIndex, routerType.Version)
 					if c.ScaleIndex >= 14 {
-						clone.Version = "v1.5.4"
+						clone.Version = "v1.6.21"
 					}
 				}
 				return nil

zititest/models/links-test/validation.go

@@ -21,11 +21,13 @@ import (
 	"errors"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/michaelquigley/pfxlog"
 	"github.com/openziti/channel/v5"
 	"github.com/openziti/channel/v5/protobufs"
+	"github.com/openziti/fablab/kernel/lib/tui"
 	"github.com/openziti/fablab/kernel/model"
 	"github.com/openziti/ziti/v2/common/pb/mgmt_pb"
 	"github.com/openziti/ziti/v2/controller/rest_client/link"
@@ -54,10 +56,9 @@ func sowChaos(run model.Run) error {
 func validateLinks(run model.Run) error {
 	ctrls := run.GetModel().SelectComponents(".ctrl")
 	errC := make(chan error, len(ctrls))
-	deadline := time.Now().Add(15 * time.Minute)
 	for _, ctrl := range ctrls {
 		ctrlComponent := ctrl
-		go validateLinksForCtrlWithChan(run, ctrlComponent, deadline, errC)
+		go validateLinksForCtrlWithChan(run, ctrlComponent, errC)
 	}
 
 	for i := 0; i < len(ctrls); i++ {
@@ -70,29 +71,64 @@ func validateLinks(run model.Run) error {
 	return nil
 }
 
-func validateLinksForCtrlWithChan(run model.Run, c *model.Component, deadline time.Time, errC chan<- error) {
-	errC <- validateLinksForCtrl(run, c, deadline)
+func validateLinksForCtrlWithChan(run model.Run, c *model.Component, errC chan<- error) {
+	errC <- validateLinksForCtrl(run, c)
 }
 
-func validateLinksForCtrl(run model.Run, c *model.Component, deadline time.Time) error {
+// expectedLinkCount is the number of links in a fully-converged mesh: exactly
+// one link per unordered router pair (links are single-dialer), C(400,2) =
+// 400*399/2.
+const expectedLinkCount = 79800
+
+// validationTimeout is the per-controller budget to reach a fully-converged,
+// validated state. maxHostUnreachableExtension is how much of that budget can be
+// reclaimed when the controller's host is unreachable, so an infra/network blip
+// (as opposed to slow convergence) does not fail the run.
+const (
+	validationTimeout           = 15 * time.Minute
+	maxHostUnreachableExtension = 10 * time.Minute
+)
+
+func validateLinksForCtrl(run model.Run, c *model.Component) error {
 	clients, err := chaos.EnsureLoggedIntoCtrl(run, c, time.Minute)
 	if err != nil {
 		return err
 	}
 
+	deadline := chaos.NewConvergenceDeadline(validationTimeout, maxHostUnreachableExtension)
 	allLinksPresent := false
 	start := time.Now()
 
-	logger := pfxlog.Logger().WithField("ctrl", c.Id)
+	logger := tui.ValidationLogger().WithField("ctrl", c.Id)
 	var lastLog time.Time
-	for time.Now().Before(deadline) && !allLinksPresent {
+	for !deadline.Expired() && !allLinksPresent {
 		linkCount, err := getLinkCount(clients)
 		if err != nil {
-			return nil
+			// A failure to reach the controller is only a convergence problem if
+			// the host is up; if the host itself is unreachable it is infra, so
+			// extend the deadline rather than charge it to the budget. Either way
+			// keep retrying (re-login best-effort) until the deadline; a login
+			// error no longer fails the run outright, since a transient blip
+			// would otherwise look like non-convergence.
+			if deadline.ExtendForUnreachableHost(run, c) {
+				logger.Warn("controller host unreachable; extending deadline (infra, not convergence)")
+			}
+			logger.WithError(err).Warn("failed to get link count, retrying")
+			time.Sleep(5 * time.Second)
+			if newClients, loginErr := chaos.EnsureLoggedIntoCtrl(run, c, time.Minute); loginErr == nil {
+				clients = newClients
+			} else {
+				logger.WithError(loginErr).Warn("failed to log in to controller, will retry")
+			}
+			continue
 		}
-		if linkCount == 79800 {
+		if linkCount == expectedLinkCount {
 			allLinksPresent = true
 		} else {
+			// The poll reached the controller, so this is reachable, slow-converging
+			// time that must be charged to the budget. Record it so the deadline's
+			// unreachable-time accounting isn't credited for this interval.
+			deadline.Observe(true)
 			time.Sleep(5 * time.Second)
 		}
 		if time.Since(lastLog) > time.Minute {
@@ -106,21 +142,28 @@ func validateLinksForCtrl(run model.Run, c *model.Component, deadline time.Time)
 	} else {
 		linkCount, _ := getLinkCount(clients)
 		logLinkDiagnostics(logger, clients, linkCount)
-		return fmt.Errorf("fail to reach expected link count of 79800 on controller %v (got %v)", c.Id, linkCount)
+		return fmt.Errorf("fail to reach expected link count of %d on controller %v (got %v)", expectedLinkCount, c.Id, linkCount)
 	}
 
 	for {
-		count, err := validateRouterLinks(c.Id, clients)
+		linkErrs, err := validateRouterLinks(c.Id, clients)
 		if err == nil {
 			logger.Infof("link validation success: elapsed time: %v", time.Since(start))
 			return nil
 		}
 
-		if time.Now().After(deadline) {
+		// As above: don't let an unreachable-host (infra) outage count against the
+		// deadline; a real validation error (host reachable, state wrong) still
+		// fails once the deadline passes.
+		deadline.ExtendForUnreachableHost(run, c)
+		if deadline.Expired() {
 			return err
 		}
 
-		logger.Infof("current link errors: %v, elapsed time: %v", count, time.Since(start))
+		// Log why the pass did not succeed, not just the counts: an unreachable router reports zero link
+		// errors, so counts alone leave a retrying-until-deadline run with no visible cause.
+		logger.Infof("validation not yet complete: %v (link errors: %v, elapsed time: %v)",
+			err, linkErrs, time.Since(start))
 		time.Sleep(15 * time.Second)
 	}
 }
@@ -143,7 +186,7 @@ func getLinkCount(clients *zitirest.Clients) (int64, error) {
 }
 
 func validateRouterLinks(id string, clients *zitirest.Clients) (int, error) {
-	logger := pfxlog.Logger().WithField("ctrl", id)
+	logger := tui.ValidationLogger().WithField("ctrl", id)
 
 	closeNotify := make(chan struct{})
 	eventNotify := make(chan *mgmt_pb.RouterLinkDetails, 1)
@@ -193,15 +236,22 @@ func validateRouterLinks(id string, clients *zitirest.Clients) (int, error) {
 	expected := response.RouterCount
 
 	invalid := 0
+	var unreachable []string
 	for expected > 0 {
 		select {
 		case <-closeNotify:
-			fmt.Printf("channel closed, exiting")
 			return 0, errors.New("unexpected close of mgmt channel")
 		case routerDetail := <-eventNotify:
+			expected--
 			if !routerDetail.ValidateSuccess {
-				return invalid, fmt.Errorf("error: unable to validate router %s (%s) on controller %s (%s)",
-					routerDetail.RouterId, routerDetail.RouterName, id, routerDetail.Message)
+				// The controller could not reach this router, so its links are unknown rather than wrong.
+				// Collect it and keep draining the pass: during chaos a router is routinely restarting or
+				// mid-reconnect, and failing here would both abandon the remaining routers' results and
+				// turn a momentary gap into a failed run. The caller retries until its deadline, so a
+				// router that never becomes reachable still fails the run, named below.
+				unreachable = append(unreachable,
+					fmt.Sprintf("%s (%s): %s", routerDetail.RouterName, routerDetail.RouterId, routerDetail.Message))
+				continue
 			}
 			for _, linkDetail := range routerDetail.LinkDetails {
 				if !linkDetail.IsValid {
@@ -213,18 +263,34 @@ func validateRouterLinks(id string, clients *zitirest.Clients) (int, error) {
 						linkDetail.Dialed, linkDetail.Messages)
 				}
 			}
-			expected--
 		}
 	}
-	if invalid == 0 {
-		logger.Infof("link validation of %v routers successful", response.RouterCount)
-		return invalid, nil
+
+	// An inconsistent link is a real failure and is reported ahead of unreachability, which may just be
+	// churn that has not settled yet.
+	if invalid > 0 {
+		return invalid, fmt.Errorf("invalid links found")
 	}
-	return invalid, fmt.Errorf("invalid links found")
+	if len(unreachable) > 0 {
+		return invalid, fmt.Errorf("%d of %d routers could not be validated on controller %s: %s",
+			len(unreachable), response.RouterCount, id, describeUnreachable(unreachable))
+	}
+	logger.Infof("link validation of %v routers successful", response.RouterCount)
+	return invalid, nil
+}
+
+// describeUnreachable renders the routers or components a controller could not reach, naming the first few
+// so a failure says where to look, without embedding hundreds of entries in one error.
+func describeUnreachable(entries []string) string {
+	const maxNamed = 5
+	if len(entries) <= maxNamed {
+		return strings.Join(entries, "; ")
+	}
+	return fmt.Sprintf("%s; and %d more", strings.Join(entries[:maxNamed], "; "), len(entries)-maxNamed)
 }
 
 func logLinkDiagnostics(logger *logrus.Entry, clients *zitirest.Clients, linkCount int64) {
-	logger.Infof("link count mismatch: expected 79800, got %v, fetching diagnostics", linkCount)
+	logger.Infof("link count mismatch: expected %d, got %v, fetching diagnostics", expectedLinkCount, linkCount)
 
 	ctx, cancelF := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancelF()
@@ -239,68 +305,66 @@ func logLinkDiagnostics(logger *logrus.Entry, clients *zitirest.Clients, linkCou
 		return
 	}
 
-	// Build a set of directed src->dst pairs from existing links, and track per-router link counts
-	type directedPair struct {
-		src, dst string
-	}
-
-	existingLinks := map[directedPair][]*rest_model.LinkDetail{}
-	routerLinkCount := map[string]int{}
+	// Links are single-dialer: each unordered router pair should have exactly one
+	// link, in whichever direction the dialer chose. Normalize each link to an
+	// unordered pair and look for pairs with the wrong number of links. A directed
+	// view (expecting src->dst for every ordered pair) is wrong here — it flags
+	// every normal one-directional link as "missing" and every router as off-count.
+	type pairKey struct{ a, b string } // a <= b
+	linksByPair := map[pairKey][]*rest_model.LinkDetail{}
 	routerIds := map[string]struct{}{}
 
 	for _, l := range result.Payload.Data {
 		src := l.SourceRouter.ID
 		dst := l.DestRouter.ID
-		key := directedPair{src: src, dst: dst}
-		existingLinks[key] = append(existingLinks[key], l)
-		routerLinkCount[src]++
 		routerIds[src] = struct{}{}
 		routerIds[dst] = struct{}{}
+		k := pairKey{a: src, b: dst}
+		if k.a > k.b {
+			k.a, k.b = k.b, k.a
+		}
+		linksByPair[k] = append(linksByPair[k], l)
 	}
 
-	// Log duplicate links (same src->dst pair with multiple links)
-	for pair, links := range existingLinks {
+	// Duplicate pairs: more than one link for the same unordered pair. This is the
+	// real anomaly when the count is over expected (e.g., both directions present
+	// because dedup didn't converge).
+	dupPairs := 0
+	for k, links := range linksByPair {
 		if len(links) > 1 {
-			logger.Infof("DUPLICATE: %v -> %v has %d links:", pair.src, pair.dst, len(links))
+			dupPairs++
+			logger.Infof("DUPLICATE pair %v <-> %v has %d links:", k.a, k.b, len(links))
 			for _, l := range links {
-				logger.Infof("  link %v: state=%v iteration=%v", *l.ID, *l.State, *l.Iteration)
+				logger.Infof("  link %v: %v -> %v state=%v iteration=%v",
+					*l.ID, l.SourceRouter.ID, l.DestRouter.ID, *l.State, *l.Iteration)
 			}
 		}
 	}
 
-	// Find routers with fewer links than expected (should be numRouters-1 links sourced from each)
-	expectedPerRouter := len(routerIds) - 1
-	var routersWithMissing []string
+	// Missing pairs: unordered router pairs with no link in either direction. This
+	// is the real anomaly when the count is under expected.
+	ids := make([]string, 0, len(routerIds))
 	for id := range routerIds {
-		count := routerLinkCount[id]
-		if count != expectedPerRouter {
-			routersWithMissing = append(routersWithMissing, id)
-			logger.Infof("router %v has %d source links (expected %d)", id, count, expectedPerRouter)
-		}
+		ids = append(ids, id)
 	}
-
-	sort.Strings(routersWithMissing)
-
-	// For routers with missing links, find which dest routers are missing
-	for _, srcId := range routersWithMissing {
-		for dstId := range routerIds {
-			if srcId == dstId {
-				continue
-			}
-			key := directedPair{src: srcId, dst: dstId}
-			if _, ok := existingLinks[key]; !ok {
-				// check if the reverse link exists
-				reverseKey := directedPair{src: dstId, dst: srcId}
-				reverseLinks := existingLinks[reverseKey]
-				reverseInfo := "no"
-				if len(reverseLinks) > 0 {
-					reverseInfo = fmt.Sprintf("yes (id=%v, iteration=%v)", *reverseLinks[0].ID, *reverseLinks[0].Iteration)
+	sort.Strings(ids)
+	missingPairs := 0
+	const missingLogCap = 50
+	for i := 0; i < len(ids); i++ {
+		for j := i + 1; j < len(ids); j++ {
+			k := pairKey{a: ids[i], b: ids[j]}
+			if _, ok := linksByPair[k]; !ok {
+				missingPairs++
+				if missingPairs <= missingLogCap {
+					logger.Infof("MISSING pair: %v <-> %v (no link in either direction)", ids[i], ids[j])
 				}
-				logger.Infof("MISSING link: %v -> %v (reverse exists: %v)", srcId, dstId, reverseInfo)
 			}
 		}
 	}
+	if missingPairs > missingLogCap {
+		logger.Infof("... and %d more missing pairs (log output capped at %d)", missingPairs-missingLogCap, missingLogCap)
+	}
 
-	logger.Infof("total links: %v, routers: %v, directed pairs: %v, routers with wrong count: %v",
-		len(result.Payload.Data), len(routerIds), len(existingLinks), len(routersWithMissing))
+	logger.Infof("total links: %v, routers: %v, unique pairs: %v, duplicate pairs: %v, missing pairs: %v",
+		len(result.Payload.Data), len(routerIds), len(linksByPair), dupPairs, missingPairs)
 }

zititest/models/oidc-auth-test/configs/router.yml.tmpl

@@ -41,7 +41,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/private-ctrl-test/configs/router-lifecycle-dialable.yml.tmpl

@@ -52,7 +52,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
   - binding: edge

zititest/models/private-ctrl-test/configs/router-lifecycle.yml.tmpl

@@ -45,7 +45,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
   - binding: edge

zititest/models/private-ctrl-test/configs/router.yml.tmpl

@@ -48,7 +48,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{- if .Component.HasTag "tunneler"}}

zititest/models/router-data-model-test/configs/router.yml.tmpl

@@ -41,7 +41,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/sdk-hosting-test/configs/router.yml.tmpl

@@ -41,7 +41,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/smoke/configs/router.yml.tmpl

@@ -51,7 +51,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 {{end}}
 
 listeners:

zititest/models/upgrade-test/configs/router.yml.tmpl

@@ -45,7 +45,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/models/zrok-test/configs/router.yml.tmpl

@@ -40,7 +40,7 @@ link:
   dialers:
     - binding:          transport
       options:
-        connectTimeout: 30s
+        connectTimeoutMs: 30000
 
 listeners:
 {{if .Component.HasTag "tunneler"}}

zititest/zitilab/component_ziti_edge_tunnel.go

@@ -40,7 +40,8 @@ type ZitiEdgeTunnelType struct {
 	Version     string
 	ZitiVersion string
 	LocalPath   string
-	// LogVerbosity is the ziti-edge-tunnel ZITI_LOG spec (e.g. "2;bind.c=6"); empty leaves it unset.
+	// LogVerbosity is the ziti-edge-tunnel ZITI_LOG spec; empty leaves it unset. It is inserted into
+	// the start command unescaped, so shell-quote a spec containing metacharacters (e.g. "'2;bind.c=6'").
 	LogVerbosity   string
 	VerbosityLevel uint16
 	ConfigPathF    func(c *model.Component) string