Additional fixes and tweaks

router/link/link_events.go

@@ -237,6 +237,7 @@ func (self *updateLinkStatusForLink) Handle(registry *linkRegistryImpl) {
 		state.connectedCount++
 		state.retryDelay = time.Duration(0)
 		state.ctrlsNotified = false
+		registry.triggerNotify()
 	}
 
 	if state.status == StatusLinkFailed {

router/link/link_registry.go

@@ -53,6 +53,7 @@ func NewLinkRegistry(routerEnv Env) xlink.Registry {
 		env:            routerEnv,
 		destinations:   map[string]*linkDest{},
 		linkStateQueue: &linkStateHeap{},
+		triggerNotifyC: make(chan struct{}, 1),
 	}
 
 	go result.run()
@@ -71,6 +72,7 @@ type linkRegistryImpl struct {
 	destinations     map[string]*linkDest
 	linkStateQueue   *linkStateHeap
 	events           chan event
+	triggerNotifyC   chan struct{}
 	notifyInProgress atomic.Bool
 }
 
@@ -365,6 +367,8 @@ func (self *linkRegistryImpl) run() {
 		select {
 		case evt := <-self.events:
 			evt.Handle(self)
+		case <-self.triggerNotifyC:
+			self.notifyControllersOfLinks()
 		case <-queueCheckTicker.C:
 			self.evaluateLinkStateQueue()
 			self.notifyControllersOfLinks()
@@ -376,6 +380,13 @@ func (self *linkRegistryImpl) run() {
 	}
 }
 
+func (self *linkRegistryImpl) triggerNotify() {
+	select {
+	case self.triggerNotifyC <- struct{}{}:
+	default:
+	}
+}
+
 func (self *linkRegistryImpl) evaluateLinkStateQueue() {
 	now := time.Now()
 	for len(*self.linkStateQueue) > 0 {

tests/control.go

@@ -4,7 +4,10 @@ import (
 	"github.com/openziti/channel/v2"
 	"github.com/openziti/foundation/v2/versions"
 	"github.com/openziti/transport/v2"
+	"github.com/openziti/ziti/common/capabilities"
+	"github.com/openziti/ziti/common/pb/ctrl_pb"
 	"github.com/openziti/ziti/controller"
+	"math/big"
 )
 
 func (ctx *FabricTestContext) NewControlChannelListener() channel.UnderlayListener {
@@ -14,8 +17,13 @@ func (ctx *FabricTestContext) NewControlChannelListener() channel.UnderlayListen
 
 	versionHeader, err := versions.StdVersionEncDec.Encode(versions.NewDefaultVersionProvider().AsVersionInfo())
 	ctx.Req.NoError(err)
+
+	capabilityMask := &big.Int{}
+	capabilityMask.SetBit(capabilityMask, capabilities.ControllerCreateTerminatorV2, 1)
+	capabilityMask.SetBit(capabilityMask, capabilities.ControllerSingleRouterLinkSource, 1)
 	headers := map[int32][]byte{
-		channel.HelloVersionHeader: versionHeader,
+		channel.HelloVersionHeader:                       versionHeader,
+		int32(ctrl_pb.ControlHeaders_CapabilitiesHeader): capabilityMask.Bytes(),
 	}
 
 	ctrlChannelListenerConfig := channel.ListenerConfig{

tests/link_test.go

@@ -223,15 +223,16 @@ func Test_DuplicateLinkWithLinkCloseDialer(t *testing.T) {
 	ctrlListener := ctx.NewControlChannelListener()
 	router1 := ctx.startRouter(1)
 
-	router1cc, linkCheck1 := testutil.StartLinkTest("router-1", ctrlListener, ctx.Req)
+	linkChecker := testutil.NewLinkChecker(ctx.Req)
+	router1cc := testutil.StartLinkTest(linkChecker, "router-1", ctrlListener, ctx.Req)
 
 	router1Listeners := &ctrl_pb.Listeners{}
 	if val, found := router1cc.Underlay().Headers()[int32(ctrl_pb.ControlHeaders_ListenersHeader)]; found {
 		ctx.Req.NoError(proto.Unmarshal(val, router1Listeners))
 	}
 
 	router2 := ctx.startRouter(2)
-	router2cc, linkCheck2 := testutil.StartLinkTest("router-2", ctrlListener, ctx.Req)
+	router2cc := testutil.StartLinkTest(linkChecker, "router-2", ctrlListener, ctx.Req)
 
 	router2Listeners := &ctrl_pb.Listeners{}
 	if val, found := router1cc.Underlay().Headers()[int32(ctrl_pb.ControlHeaders_ListenersHeader)]; found {
@@ -266,32 +267,31 @@ func Test_DuplicateLinkWithLinkCloseDialer(t *testing.T) {
 
 	time.Sleep(time.Second)
 
-	linkCheck1.RequireNoErrors()
-	link1 := linkCheck1.RequireOneActiveLink()
+	linkChecker.RequireNoErrors()
+	link1 := linkChecker.RequireOneActiveLink()
 
-	linkCheck2.RequireNoErrors()
-	link2 := linkCheck1.RequireOneActiveLink()
+	linkChecker.RequireNoErrors()
+	link2 := linkChecker.RequireOneActiveLink()
 
 	ctx.Req.Equal(link1.Id, link2.Id)
 
 	// Test closing control ch to router 1. On reconnect the existing link should get reported
 	ctx.Req.NoError(router1cc.Close())
-	_, linkCheck1 = testutil.StartLinkTest("router-1", ctrlListener, ctx.Req)
+	_ = testutil.StartLinkTest(linkChecker, "router-1", ctrlListener, ctx.Req)
 
 	time.Sleep(time.Second)
-
-	linkCheck1.RequireNoErrors()
-	link1 = linkCheck1.RequireOneActiveLink()
+	linkChecker.RequireNoErrors()
+	link1 = linkChecker.RequireOneActiveLink()
 	ctx.Req.Equal(link1.Id, link2.Id)
 
 	// Test closing control ch to router 2. On reconnect the existing link should get reported
 	ctx.Req.NoError(router2cc.Close())
-	_, linkCheck2 = testutil.StartLinkTest("router-2", ctrlListener, ctx.Req)
+	_ = testutil.StartLinkTest(linkChecker, "router-2", ctrlListener, ctx.Req)
 
 	time.Sleep(time.Second)
 
-	linkCheck2.RequireNoErrors()
-	link2 = linkCheck2.RequireOneActiveLink()
+	linkChecker.RequireNoErrors()
+	link2 = linkChecker.RequireOneActiveLink()
 	ctx.Req.Equal(link1.Id, link2.Id)
 
 	// restart router 1
@@ -302,10 +302,10 @@ func Test_DuplicateLinkWithLinkCloseDialer(t *testing.T) {
 		ctx.Req.NoError(router1.Shutdown())
 	}()
 
-	router1cc, linkCheck1 = testutil.StartLinkTest("router-1", ctrlListener, ctx.Req)
+	router1cc = testutil.StartLinkTest(linkChecker, "router-1", ctrlListener, ctx.Req)
 	ctx.Req.NoError(protobufs.MarshalTyped(peerUpdates2).WithTimeout(time.Second).SendAndWaitForWire(router1cc))
 
-	linkCheck1.RequireNoErrors()
+	linkChecker.RequireNoErrors()
 
 	//time.Sleep(time.Minute)
 	//

tests/testutil/linkschecker.go

@@ -1,6 +1,7 @@
 package testutil
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -15,9 +16,11 @@ import (
 )
 
 type TestLink struct {
-	Id     string
-	Dest   string
-	Failed bool
+	Id         string
+	Src        string
+	Dest       string
+	FaultCount int
+	Valid      bool
 }
 
 type LinkStateChecker struct {
@@ -34,7 +37,7 @@ func (self *LinkStateChecker) reportError(err error) {
 	}
 }
 
-func (self *LinkStateChecker) HandleLink(msg *channel.Message, _ channel.Channel) {
+func (self *LinkStateChecker) HandleLink(msg *channel.Message, ch channel.Channel) {
 	self.Lock()
 	defer self.Unlock()
 
@@ -44,8 +47,22 @@ func (self *LinkStateChecker) HandleLink(msg *channel.Message, _ channel.Channel
 	}
 
 	for _, link := range routerLinks.Links {
-		self.links[link.Id] = &TestLink{
-			Id: link.Id,
+		testLink, ok := self.links[link.Id]
+		if !ok {
+			self.links[link.Id] = &TestLink{
+				Id:    link.Id,
+				Src:   ch.Id(),
+				Dest:  link.DestRouterId,
+				Valid: true,
+			}
+		} else {
+			if testLink.Src != ch.Id() {
+				self.reportError(fmt.Errorf("source router change for link %v => %v", testLink.Src, ch.Id()))
+			}
+			if testLink.Dest != link.DestRouterId {
+				self.reportError(fmt.Errorf("dest router change for link %v => %v", testLink.Dest, link.DestRouterId))
+			}
+			testLink.Valid = true
 		}
 	}
 }
@@ -64,7 +81,8 @@ func (self *LinkStateChecker) HandleFault(msg *channel.Message, _ channel.Channe
 
 	if fault.Subject == ctrl_pb.FaultSubject_LinkFault || fault.Subject == ctrl_pb.FaultSubject_LinkDuplicate {
 		if link, found := self.links[fault.Id]; found {
-			link.Failed = true
+			link.FaultCount++
+			link.Valid = false
 		} else {
 			self.reportError(errors.Errorf("no link with Id %s found", fault.Id))
 		}
@@ -108,7 +126,7 @@ func (self *LinkStateChecker) RequireOneActiveLink() *TestLink {
 	var activeLink *TestLink
 
 	for _, link := range self.links {
-		if !link.Failed {
+		if link.Valid {
 			self.req.Nil(activeLink, "more than one active link found")
 			activeLink = link
 		}
@@ -117,13 +135,16 @@ func (self *LinkStateChecker) RequireOneActiveLink() *TestLink {
 	return activeLink
 }
 
-func StartLinkTest(id string, uf channel.UnderlayFactory, assertions *require.Assertions) (channel.Channel, *LinkStateChecker) {
+func NewLinkChecker(assertions *require.Assertions) *LinkStateChecker {
 	checker := &LinkStateChecker{
 		errorC: make(chan error, 4),
 		links:  map[string]*TestLink{},
 		req:    assertions,
 	}
+	return checker
+}
 
+func StartLinkTest(checker *LinkStateChecker, id string, uf channel.UnderlayFactory, assertions *require.Assertions) channel.Channel {
 	bindHandler := func(binding channel.Binding) error {
 		binding.AddReceiveHandlerF(channel.AnyContentType, checker.HandleOther)
 		binding.AddReceiveHandlerF(int32(ctrl_pb.ContentType_VerifyRouterType), func(msg *channel.Message, ch channel.Channel) {
@@ -137,5 +158,5 @@ func StartLinkTest(id string, uf channel.UnderlayFactory, assertions *require.As
 	timeoutUF := NewTimeoutUnderlayFactory(uf, 2*time.Second)
 	ch, err := channel.NewChannel(id, timeoutUF, channel.BindHandlerF(bindHandler), channel.DefaultOptions())
 	assertions.NoError(err)
-	return ch, checker
+	return ch
 }