lib/connections: React to listeners going up and down faster (#6590)

lib/connections/quic_listen.go

@@ -102,7 +102,9 @@ func (t *quicListener) serve(ctx context.Context) error {
 		l.Infoln("Listen (BEP/quic):", err)
 		return err
 	}
+	t.notifyAddressesChanged(t)
 	defer listener.Close()
+	defer t.clearAddresses(t)
 
 	l.Infof("QUIC listener (%v) starting", packetConn.LocalAddr())
 	defer l.Infof("QUIC listener (%v) shutting down", packetConn.LocalAddr())

lib/connections/relay_listen.go

@@ -57,10 +57,12 @@ func (t *relayListener) serve(ctx context.Context) error {
 	defer clnt.Stop()
 	t.mut.Unlock()
 
-	oldURI := clnt.URI()
+	// Start with nil, so that we send a addresses changed notification as soon as we connect somewhere.
+	var oldURI *url.URL
 
 	l.Infof("Relay listener (%v) starting", t)
 	defer l.Infof("Relay listener (%v) shutting down", t)
+	defer t.clearAddresses(t)
 
 	for {
 		select {

lib/connections/service.go

@@ -565,11 +565,11 @@ func (s *service) createListener(factory listenerFactory, uri *url.URL) bool {
 	return true
 }
 
-func (s *service) logListenAddressesChangedEvent(l genericListener) {
+func (s *service) logListenAddressesChangedEvent(l ListenerAddresses) {
 	s.evLogger.Log(events.ListenAddressesChanged, map[string]interface{}{
-		"address": l.URI(),
-		"lan":     l.LANAddresses(),
-		"wan":     l.WANAddresses(),
+		"address": l.URI,
+		"lan":     l.LANAddresses,
+		"wan":     l.WANAddresses,
 	})
 }
 

lib/connections/structs.go

@@ -174,6 +174,12 @@ type listenerFactory interface {
 	Valid(config.Configuration) error
 }
 
+type ListenerAddresses struct {
+	URI          *url.URL
+	WANAddresses []*url.URL
+	LANAddresses []*url.URL
+}
+
 type genericListener interface {
 	Serve()
 	Stop()
@@ -188,7 +194,7 @@ type genericListener interface {
 	WANAddresses() []*url.URL
 	LANAddresses() []*url.URL
 	Error() error
-	OnAddressesChanged(func(genericListener))
+	OnAddressesChanged(func(ListenerAddresses))
 	String() string
 	Factory() listenerFactory
 	NATType() string
@@ -203,14 +209,28 @@ type Model interface {
 }
 
 type onAddressesChangedNotifier struct {
-	callbacks []func(genericListener)
+	callbacks []func(ListenerAddresses)
 }
 
-func (o *onAddressesChangedNotifier) OnAddressesChanged(callback func(genericListener)) {
+func (o *onAddressesChangedNotifier) OnAddressesChanged(callback func(ListenerAddresses)) {
 	o.callbacks = append(o.callbacks, callback)
 }
 
 func (o *onAddressesChangedNotifier) notifyAddressesChanged(l genericListener) {
+	o.notifyAddresses(ListenerAddresses{
+		URI:          l.URI(),
+		WANAddresses: l.WANAddresses(),
+		LANAddresses: l.LANAddresses(),
+	})
+}
+
+func (o *onAddressesChangedNotifier) clearAddresses(l genericListener) {
+	o.notifyAddresses(ListenerAddresses{
+		URI: l.URI(),
+	})
+}
+
+func (o *onAddressesChangedNotifier) notifyAddresses(l ListenerAddresses) {
 	for _, callback := range o.callbacks {
 		callback(l)
 	}

lib/connections/tcp_listen.go

@@ -55,7 +55,9 @@ func (t *tcpListener) serve(ctx context.Context) error {
 		l.Infoln("Listen (BEP/tcp):", err)
 		return err
 	}
+	t.notifyAddressesChanged(t)
 	defer listener.Close()
+	defer t.clearAddresses(t)
 
 	l.Infof("TCP listener (%v) starting", listener.Addr())
 	defer l.Infof("TCP listener (%v) shutting down", listener.Addr())

lib/discover/global.go

@@ -46,9 +46,10 @@ type httpClient interface {
 }
 
 const (
-	defaultReannounceInterval  = 30 * time.Minute
-	announceErrorRetryInterval = 5 * time.Minute
-	requestTimeout             = 5 * time.Second
+	defaultReannounceInterval             = 30 * time.Minute
+	announceErrorRetryInterval            = 5 * time.Minute
+	requestTimeout                        = 5 * time.Second
+	maxAddressChangesBetweenAnnouncements = 10
 )
 
 type announcement struct {
@@ -197,20 +198,33 @@ func (c *globalClient) serve(ctx context.Context) {
 		return
 	}
 
-	timer := time.NewTimer(0)
+	timer := time.NewTimer(5 * time.Second)
 	defer timer.Stop()
 
 	eventSub := c.evLogger.Subscribe(events.ListenAddressesChanged)
 	defer eventSub.Unsubscribe()
 
+	timerResetCount := 0
+
 	for {
 		select {
 		case <-eventSub.C():
-			// Defer announcement by 2 seconds, essentially debouncing
-			// if we have a stream of events incoming in quick succession.
-			timer.Reset(2 * time.Second)
-
+			if timerResetCount < maxAddressChangesBetweenAnnouncements {
+				// Defer announcement by 2 seconds, essentially debouncing
+				// if we have a stream of events incoming in quick succession.
+				timer.Reset(2 * time.Second)
+			} else if timerResetCount == maxAddressChangesBetweenAnnouncements {
+				// Yet only do it if we haven't had to reset maxAddressChangesBetweenAnnouncements times in a row,
+				// so if something is flip-flopping within 2 seconds, we don't end up in a permanent reset loop.
+				l.Warnf("Detected a flip-flopping listener")
+				c.setError(errors.New("flip flopping listener"))
+				// Incrementing the count above 10 will prevent us from warning or setting the error again
+				// It will also suppress event based resets until we've had a proper round after announceErrorRetryInterval
+				timer.Reset(announceErrorRetryInterval)
+			}
+			timerResetCount++
 		case <-timer.C:
+			timerResetCount = 0
 			c.sendAnnouncement(ctx, timer)
 
 		case <-ctx.Done():
@@ -237,7 +251,7 @@ func (c *globalClient) sendAnnouncement(ctx context.Context, timer *time.Timer)
 	// The marshal doesn't fail, I promise.
 	postData, _ := json.Marshal(ann)
 
-	l.Debugf("Announcement: %s", postData)
+	l.Debugf("Announcement: %v", ann)
 
 	resp, err := c.announceClient.Post(ctx, c.server, "application/json", bytes.NewReader(postData))
 	if err != nil {