Stop a router dialing a controller it is already connected to. For #4618

- keeps one dial per controller endpoint, so an endpoint removed and added back while its dial is under way doesn't start a second dial to the same controller
- stops dialing an endpoint that reaches an already connected controller, taking the refusal from the bind rather than from the dialer's error, and dials it again once that controller's connection is removed
- refuses to register a controller reached on an endpoint that was removed while it was dialed, serializing registration with endpoint removal
- wraps the control channel connect error with %w so its cause stays in the chain
- logs the endpoint being added or removed rather than the whole set
- tests the single dial, the stopped duplicate and its redial, and the refused registration

router/env/ctrls.go

@@ -94,11 +94,32 @@ func NewNetworkControllers(defaultRequestTimeout time.Duration, dialer CtrlDiale
 		heartbeatOptions:      heartbeatOptions,
 		defaultRequestTimeout: defaultRequestTimeout,
 		ctrlEndpoints:         cmap.New[struct{}](),
+		dialing:               map[string]struct{}{},
+		duplicates:            map[string]string{},
 	}
 }
 
+// errCtrlEndpointRemoved ends the dial of an endpoint that is no longer wanted.
+var errCtrlEndpointRemoved = errors.New("controller removed before connection established")
+
+// duplicateCtrlError reports that an endpoint reached a controller the router is already connected to.
+type duplicateCtrlError struct {
+	ctrlId string
+}
+
+func (self *duplicateCtrlError) Error() string {
+	return fmt.Sprintf("duplicate channel with id %v", self.ctrlId)
+}
+
 type networkControllers struct {
-	lock                  sync.Mutex
+	// lock serializes changes to the wanted endpoints with registering a dialed controller, and guards
+	// dialing and duplicates.
+	lock sync.Mutex
+	// dialing holds the endpoints a dial is under way for.
+	dialing map[string]struct{}
+	// duplicates maps an endpoint whose dial reached an already connected controller to that
+	// controller's id. The endpoint is dialed again once that controller is no longer connected.
+	duplicates            map[string]string
 	ctrlDialer            CtrlDialer
 	heartbeatOptions      *HeartbeatOptions
 	defaultRequestTimeout time.Duration
@@ -150,18 +171,28 @@ func (self *networkControllers) UpdateControllerEndpoints(addresses []string) bo
 			delete(endpoints, endpoint)
 		} else {
 			// existing endpoint is no longer valid, close and remove it
-			log.WithField("endpoint", endpoints).Info("removing old ctrl endpoint")
+			log.WithField("endpoint", endpoint).Info("removing old ctrl endpoint")
 			changed = true
+			delete(self.duplicates, endpoint)
 			self.CloseAndRemoveByAddress(endpoint)
 		}
 	}
 
 	for endpoint := range endpoints {
-		log.WithField("endpoint", endpoints).Info("adding new ctrl endpoint")
+		log.WithField("endpoint", endpoint).Info("adding new ctrl endpoint")
 		changed = true
 		self.connectToControllerWithBackoff(endpoint)
 	}
 
+	for endpoint, ctrlId := range self.duplicates {
+		if ctrl := self.ctrls.Get(ctrlId); ctrl == nil || ctrl.Channel().IsClosed() {
+			log.WithField("endpoint", endpoint).WithField("ctrlId", ctrlId).
+				Info("controller the endpoint duplicated is no longer connected, dialing endpoint")
+			delete(self.duplicates, endpoint)
+			self.connectToControllerWithBackoff(endpoint)
+		}
+	}
+
 	return changed
 }
 
@@ -177,6 +208,10 @@ func (self *networkControllers) UpdateLeader(leaderId string) {
 	}
 }
 
+// connectToControllerWithBackoff adds endpoint to the wanted endpoints and dials it until a controller
+// is registered on it, unless a dial for it is already under way. The dial stops once the endpoint is
+// no longer wanted, or when it reaches a controller the router is already connected to, recording the
+// endpoint in duplicates. The caller holds lock.
 func (self *networkControllers) connectToControllerWithBackoff(endpoint string) {
 	log := pfxlog.Logger().WithField("endpoint", endpoint)
 
@@ -188,11 +223,15 @@ func (self *networkControllers) connectToControllerWithBackoff(endpoint string)
 
 	self.ctrlEndpoints.Set(endpoint, struct{}{})
 
-	expBackoff := backoff.NewExponentialBackOff()
-	expBackoff.InitialInterval = 50 * time.Millisecond
-	expBackoff.MaxInterval = 5 * time.Minute
-	expBackoff.MaxElapsedTime = 100 * 365 * 24 * time.Hour
+	if _, ok := self.dialing[endpoint]; ok {
+		log.Info("connection attempts already under way")
+		return
+	}
+	self.dialing[endpoint] = struct{}{}
 
+	// bindErr is the error from registering the controller on the last attempt. The bind runs on the
+	// dialing goroutine, within ctrlDialer.
+	var bindErr error
 	bindHandler := channel.BindHandlerF(func(binding channel.Binding) error {
 		id := binding.GetChannel().Id()
 		binding.AddReceiveHandlerF(int32(edge_ctrl_pb.ContentType_CurrentIndexMessageType), func(m *channel.Message, ch channel.Channel) {
@@ -203,15 +242,24 @@ func (self *networkControllers) connectToControllerWithBackoff(endpoint string)
 			}
 		})
 
-		return self.Add(endpoint, binding.GetChannel())
+		bindErr = self.Add(endpoint, binding.GetChannel())
+		return bindErr
 	})
 
 	operation := func() error {
 		if !self.ctrlEndpoints.Has(endpoint) {
-			return backoff.Permanent(errors.New("controller removed before connection established"))
+			return backoff.Permanent(errCtrlEndpointRemoved)
 		}
+		bindErr = nil
 		err = self.ctrlDialer(addr, bindHandler)
 		if err != nil {
+			// The dialer may not keep the bind error in its chain, so a refusal that ends the dial is
+			// taken from the bind itself.
+			var permanent *backoff.PermanentError
+			if errors.As(bindErr, &permanent) {
+				log.WithError(bindErr).Info("controller connection refused, not retrying")
+				return bindErr
+			}
 			log.WithError(err).Error("unable to connect controller")
 		}
 		return err
@@ -220,14 +268,52 @@ func (self *networkControllers) connectToControllerWithBackoff(endpoint string)
 	log.Info("starting connection attempts")
 
 	go func() {
-		if err := backoff.Retry(operation, expBackoff); err != nil {
-			log.WithError(err).Error("unable to connect controller, stopping retries.")
-		} else {
-			log.Info("successfully connected to controller")
+		for {
+			expBackoff := backoff.NewExponentialBackOff()
+			expBackoff.InitialInterval = 50 * time.Millisecond
+			expBackoff.MaxInterval = 5 * time.Minute
+			expBackoff.MaxElapsedTime = 100 * 365 * 24 * time.Hour
+
+			err := backoff.Retry(operation, expBackoff)
+			if self.endDial(endpoint, err) {
+				if err != nil {
+					log.WithError(err).Error("unable to connect controller, stopping retries.")
+				} else {
+					log.Info("successfully connected to controller")
+				}
+				return
+			}
+			log.Info("endpoint still needs a connection, restarting connection attempts")
 		}
 	}()
 }
 
+// endDial records that the dial of endpoint ended with err, and reports whether it should stop. It
+// shouldn't if the endpoint was removed and then wanted again before the dial noticed, or if the
+// controller it duplicated is already gone.
+func (self *networkControllers) endDial(endpoint string, err error) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+
+	wanted := self.ctrlEndpoints.Has(endpoint)
+	if errors.Is(err, errCtrlEndpointRemoved) && wanted {
+		return false
+	}
+
+	var duplicate *duplicateCtrlError
+	if errors.As(err, &duplicate) && wanted {
+		if ctrl := self.ctrls.Get(duplicate.ctrlId); ctrl == nil || ctrl.Channel().IsClosed() {
+			return false
+		}
+		self.duplicates[endpoint] = duplicate.ctrlId
+	}
+
+	delete(self.dialing, endpoint)
+	return true
+}
+
+// Add registers ch as the controller dialed on address. It fails permanently if address is no longer
+// wanted, or if a controller with ch's id is already registered on an open channel.
 func (self *networkControllers) Add(address string, ch channel.Channel) error {
 	ctrl := newNetworkCtrl(ch, address, self.heartbeatOptions)
 
@@ -241,19 +327,33 @@ func (self *networkControllers) Add(address string, ch channel.Channel) error {
 		return errors.New("no version header provided")
 	}
 
-	if existing := self.ctrls.Get(ch.Id()); existing != nil {
-		if !existing.Channel().IsClosed() {
-			// if an existing channel exists, don't keep trying to dial one
-			return backoff.Permanent(fmt.Errorf("duplicate channel with id %v", ctrl.Channel().Id()))
-		}
+	if err := self.register(address, ctrl); err != nil {
+		return err
 	}
-	self.ctrls.Put(ch.Id(), ctrl)
 
 	self.notifyOfChange(ctrl, ControllerAdded)
 
 	return nil
 }
 
+// register adds ctrl under lock, so it is either registered before its endpoint is removed, and closed
+// by the removal, or refused after it.
+func (self *networkControllers) register(address string, ctrl *networkCtrl) error {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+
+	if !self.ctrlEndpoints.Has(address) {
+		return backoff.Permanent(errCtrlEndpointRemoved)
+	}
+
+	id := ctrl.Channel().Id()
+	if existing := self.ctrls.Get(id); existing != nil && !existing.Channel().IsClosed() {
+		return backoff.Permanent(&duplicateCtrlError{ctrlId: id})
+	}
+	self.ctrls.Put(id, ctrl)
+	return nil
+}
+
 func (self *networkControllers) NotifyOfDisconnect(ctrlId string) {
 	if ctrl := self.GetNetworkController(ctrlId); ctrl != nil {
 		self.notifyOfChange(ctrl, ControllerDisconnected)

router/env/ctrls_test.go

@@ -0,0 +1,261 @@
+/*
+	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 env
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/openziti/channel/v4"
+	"github.com/openziti/foundation/v2/versions"
+	"github.com/openziti/transport/v2"
+	"github.com/openziti/transport/v2/tls"
+	"github.com/stretchr/testify/require"
+)
+
+type testCtrlUnderlay struct {
+	channel.Underlay
+	headers map[int32][]byte
+}
+
+func (u *testCtrlUnderlay) Headers() map[int32][]byte {
+	return u.headers
+}
+
+type testCtrlChannel struct {
+	channel.Channel
+	id       string
+	closed   atomic.Bool
+	underlay *testCtrlUnderlay
+}
+
+func (c *testCtrlChannel) Id() string {
+	return c.id
+}
+
+func (c *testCtrlChannel) Label() string {
+	return c.id
+}
+
+func (c *testCtrlChannel) IsClosed() bool {
+	return c.closed.Load()
+}
+
+func (c *testCtrlChannel) Close() error {
+	c.closed.Store(true)
+	return nil
+}
+
+func (c *testCtrlChannel) Underlay() channel.Underlay {
+	return c.underlay
+}
+
+type testCtrlBinding struct {
+	channel.Binding
+	ch channel.Channel
+}
+
+func (b *testCtrlBinding) GetChannel() channel.Channel {
+	return b.ch
+}
+
+func (b *testCtrlBinding) AddReceiveHandlerF(int32, channel.ReceiveHandlerF) {}
+
+// testCtrlDialer reaches the controller each endpoint names, counting dials. A gated endpoint's first
+// dial waits until the gate is released. Bind errors are wrapped without keeping them in the chain.
+type testCtrlDialer struct {
+	t       *testing.T
+	lock    sync.Mutex
+	ctrlIds map[string]string
+	dials   map[string]int
+	gates   map[string]chan struct{}
+	started map[string]chan struct{}
+}
+
+func newTestCtrlDialer(t *testing.T) *testCtrlDialer {
+	return &testCtrlDialer{
+		t:       t,
+		ctrlIds: map[string]string{},
+		dials:   map[string]int{},
+		gates:   map[string]chan struct{}{},
+		started: map[string]chan struct{}{},
+	}
+}
+
+// gate makes endpoint's first dial wait for release, and returns a channel closed once it starts.
+func (d *testCtrlDialer) gate(endpoint string) (started <-chan struct{}, release func()) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	gateC := make(chan struct{})
+	startedC := make(chan struct{})
+	d.gates[endpoint] = gateC
+	d.started[endpoint] = startedC
+	var once sync.Once
+	release = func() {
+		once.Do(func() {
+			close(gateC)
+		})
+	}
+	d.t.Cleanup(release)
+	return startedC, release
+}
+
+func (d *testCtrlDialer) dialCount(endpoint string) int {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	return d.dials[endpoint]
+}
+
+func (d *testCtrlDialer) dial(address transport.Address, bindHandler channel.BindHandler) error {
+	endpoint := address.String()
+
+	d.lock.Lock()
+	d.dials[endpoint]++
+	ctrlId := d.ctrlIds[endpoint]
+	gateC := d.gates[endpoint]
+	startedC := d.started[endpoint]
+	delete(d.gates, endpoint)
+	d.lock.Unlock()
+
+	if gateC != nil {
+		close(startedC)
+		<-gateC
+	}
+
+	version, err := versions.StdVersionEncDec.Encode(&versions.VersionInfo{Version: "v1.6.0"})
+	if err != nil {
+		return err
+	}
+	ch := &testCtrlChannel{
+		id:       ctrlId,
+		underlay: &testCtrlUnderlay{headers: map[int32][]byte{channel.HelloVersionHeader: version}},
+	}
+
+	if err = bindHandler.BindChannel(&testCtrlBinding{ch: ch}); err != nil {
+		_ = ch.Close()
+		return fmt.Errorf("error connecting ctrl (%v)", err)
+	}
+	return nil
+}
+
+func newTestNetworkControllers(t *testing.T) (*networkControllers, *testCtrlDialer) {
+	transport.AddAddressParser(tls.AddressParser{})
+	dialer := newTestCtrlDialer(t)
+	ctrls := NewNetworkControllers(time.Minute, dialer.dial, NewDefaultHeartbeatOptions()).(*networkControllers)
+	return ctrls, dialer
+}
+
+func testCtrlEndpoint(name string) string {
+	return "tls:" + name + ":6262"
+}
+
+func (self *networkControllers) isDialing(endpoint string) bool {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	_, ok := self.dialing[endpoint]
+	return ok
+}
+
+func requireRegisteredAt(t *testing.T, ctrls *networkControllers, ctrlId string, endpoint string) {
+	require.Eventually(t, func() bool {
+		ctrl := ctrls.GetNetworkController(ctrlId)
+		return ctrl != nil && ctrl.Address() == endpoint && !ctrl.Channel().IsClosed()
+	}, 5*time.Second, time.Millisecond, "%s was not registered at %s", ctrlId, endpoint)
+}
+
+// Test_NetworkControllers_ReaddedEndpointIsNotDialedTwice: an endpoint removed and added again while its
+// dial is under way keeps that dial, rather than starting a second one that would reach the same
+// controller.
+func Test_NetworkControllers_ReaddedEndpointIsNotDialedTwice(t *testing.T) {
+	req := require.New(t)
+	ctrls, dialer := newTestNetworkControllers(t)
+	ep1, ep2 := testCtrlEndpoint("ctrl1"), testCtrlEndpoint("ctrl2")
+	dialer.ctrlIds[ep1] = "ctrl1"
+	dialer.ctrlIds[ep2] = "ctrl2"
+	started, release := dialer.gate(ep1)
+
+	ctrls.UpdateControllerEndpoints([]string{ep1, ep2})
+	select {
+	case <-started:
+	case <-time.After(5 * time.Second):
+		req.FailNow("the first dial did not start")
+	}
+
+	ctrls.UpdateControllerEndpoints([]string{ep2})
+	ctrls.UpdateControllerEndpoints([]string{ep1, ep2})
+	release()
+
+	requireRegisteredAt(t, ctrls, "ctrl1", ep1)
+	requireRegisteredAt(t, ctrls, "ctrl2", ep2)
+	req.Eventually(func() bool { return !ctrls.isDialing(ep1) }, 5*time.Second, time.Millisecond)
+	time.Sleep(200 * time.Millisecond)
+	req.Equal(1, dialer.dialCount(ep1), "an endpoint added back during its dial must not be dialed again")
+}
+
+// Test_NetworkControllers_DuplicateEndpointStopsAndIsDialedWhenItsControllerGoes: an endpoint reaching a
+// controller already connected on another endpoint stops dialing, and is dialed again once the endpoint
+// that controller was connected on is removed.
+func Test_NetworkControllers_DuplicateEndpointStopsAndIsDialedWhenItsControllerGoes(t *testing.T) {
+	req := require.New(t)
+	ctrls, dialer := newTestNetworkControllers(t)
+	viaName, viaAddress := testCtrlEndpoint("ctrl1.example.com"), testCtrlEndpoint("10.0.0.1")
+	dialer.ctrlIds[viaName] = "ctrl1"
+	dialer.ctrlIds[viaAddress] = "ctrl1"
+
+	ctrls.UpdateControllerEndpoints([]string{viaName})
+	requireRegisteredAt(t, ctrls, "ctrl1", viaName)
+
+	ctrls.UpdateControllerEndpoints([]string{viaName, viaAddress})
+	req.Eventually(func() bool { return !ctrls.isDialing(viaAddress) }, 5*time.Second, time.Millisecond,
+		"a dial reaching an already connected controller must stop")
+	time.Sleep(300 * time.Millisecond)
+	req.Equal(1, dialer.dialCount(viaAddress), "a dial reaching an already connected controller must not retry")
+	requireRegisteredAt(t, ctrls, "ctrl1", viaName)
+
+	ctrls.UpdateControllerEndpoints([]string{viaAddress})
+	requireRegisteredAt(t, ctrls, "ctrl1", viaAddress)
+	req.Equal(2, dialer.dialCount(viaAddress))
+}
+
+// Test_NetworkControllers_RemovedEndpointIsNotRegistered: a dial that completes after its endpoint was
+// removed doesn't register the controller it reached.
+func Test_NetworkControllers_RemovedEndpointIsNotRegistered(t *testing.T) {
+	req := require.New(t)
+	ctrls, dialer := newTestNetworkControllers(t)
+	ep1, ep2 := testCtrlEndpoint("ctrl1"), testCtrlEndpoint("ctrl2")
+	dialer.ctrlIds[ep1] = "ctrl1"
+	dialer.ctrlIds[ep2] = "ctrl2"
+	started, release := dialer.gate(ep1)
+
+	ctrls.UpdateControllerEndpoints([]string{ep1})
+	select {
+	case <-started:
+	case <-time.After(5 * time.Second):
+		req.FailNow("the first dial did not start")
+	}
+
+	ctrls.UpdateControllerEndpoints([]string{ep2})
+	release()
+
+	requireRegisteredAt(t, ctrls, "ctrl2", ep2)
+	req.Eventually(func() bool { return !ctrls.isDialing(ep1) }, 5*time.Second, time.Millisecond)
+	req.Nil(ctrls.GetNetworkController("ctrl1"), "a controller reached on a removed endpoint must not be registered")
+	req.Equal([]string{ep2}, ctrls.GetEndpoints())
+}

router/router.go

@@ -875,7 +875,7 @@ func (self *Router) connectToController(addr transport.Address, bindHandler chan
 	bindHandler = channel.BindHandlers(bindHandler, self.ctrlBindhandler)
 	ch, err := channel.NewChannel("ctrl", dialer, bindHandler, self.config.Ctrl.Options)
 	if err != nil {
-		return fmt.Errorf("error connecting ctrl (%v)", err)
+		return fmt.Errorf("error connecting ctrl (%w)", err)
 	}
 	channelRef.Store(ch)