all: Remove need to restart syncthing (#6883)

lib/discover/cache.go

@@ -7,41 +7,21 @@
 package discover
 
 import (
-	"context"
-	"sort"
 	stdsync "sync"
 	"time"
 
-	"github.com/syncthing/syncthing/lib/protocol"
-	"github.com/syncthing/syncthing/lib/sync"
-	"github.com/syncthing/syncthing/lib/util"
 	"github.com/thejerf/suture"
+
+	"github.com/syncthing/syncthing/lib/protocol"
 )
 
-// The CachingMux aggregates results from multiple Finders. Each Finder has
-// an associated cache time and negative cache time. The cache time sets how
-// long we cache and return successful lookup results, the negative cache
-// time sets how long we refrain from asking about the same device ID after
-// receiving a negative answer. The value of zero disables caching (positive
-// or negative).
-type CachingMux interface {
-	FinderService
-	Add(finder Finder, cacheTime, negCacheTime time.Duration)
-	ChildErrors() map[string]error
-}
-
-type cachingMux struct {
-	*suture.Supervisor
-	finders []cachedFinder
-	caches  []*cache
-	mut     sync.RWMutex
-}
-
 // A cachedFinder is a Finder with associated cache timeouts.
 type cachedFinder struct {
 	Finder
 	cacheTime    time.Duration
 	negCacheTime time.Duration
+	cache        *cache
+	token        *suture.ServiceToken
 }
 
 // An error may implement cachedError, in which case it will be interrogated
@@ -51,150 +31,6 @@ type cachedError interface {
 	CacheFor() time.Duration
 }
 
-func NewCachingMux() CachingMux {
-	return &cachingMux{
-		Supervisor: suture.New("discover.cachingMux", suture.Spec{
-			PassThroughPanics: true,
-		}),
-		mut: sync.NewRWMutex(),
-	}
-}
-
-// Add registers a new Finder, with associated cache timeouts.
-func (m *cachingMux) Add(finder Finder, cacheTime, negCacheTime time.Duration) {
-	m.mut.Lock()
-	m.finders = append(m.finders, cachedFinder{finder, cacheTime, negCacheTime})
-	m.caches = append(m.caches, newCache())
-	m.mut.Unlock()
-
-	if service, ok := finder.(suture.Service); ok {
-		m.Supervisor.Add(service)
-	}
-}
-
-// Lookup attempts to resolve the device ID using any of the added Finders,
-// while obeying the cache settings.
-func (m *cachingMux) Lookup(ctx context.Context, deviceID protocol.DeviceID) (addresses []string, err error) {
-	m.mut.RLock()
-	for i, finder := range m.finders {
-		if cacheEntry, ok := m.caches[i].Get(deviceID); ok {
-			// We have a cache entry. Lets see what it says.
-
-			if cacheEntry.found && time.Since(cacheEntry.when) < finder.cacheTime {
-				// It's a positive, valid entry. Use it.
-				l.Debugln("cached discovery entry for", deviceID, "at", finder)
-				l.Debugln("  cache:", cacheEntry)
-				addresses = append(addresses, cacheEntry.Addresses...)
-				continue
-			}
-
-			valid := time.Now().Before(cacheEntry.validUntil) || time.Since(cacheEntry.when) < finder.negCacheTime
-			if !cacheEntry.found && valid {
-				// It's a negative, valid entry. We should not make another
-				// attempt right now.
-				l.Debugln("negative cache entry for", deviceID, "at", finder, "valid until", cacheEntry.when.Add(finder.negCacheTime), "or", cacheEntry.validUntil)
-				continue
-			}
-
-			// It's expired. Ignore and continue.
-		}
-
-		// Perform the actual lookup and cache the result.
-		if addrs, err := finder.Lookup(ctx, deviceID); err == nil {
-			l.Debugln("lookup for", deviceID, "at", finder)
-			l.Debugln("  addresses:", addrs)
-			addresses = append(addresses, addrs...)
-			m.caches[i].Set(deviceID, CacheEntry{
-				Addresses: addrs,
-				when:      time.Now(),
-				found:     len(addrs) > 0,
-			})
-		} else {
-			// Lookup returned error, add a negative cache entry.
-			entry := CacheEntry{
-				when:  time.Now(),
-				found: false,
-			}
-			if err, ok := err.(cachedError); ok {
-				entry.validUntil = time.Now().Add(err.CacheFor())
-			}
-			m.caches[i].Set(deviceID, entry)
-		}
-	}
-	m.mut.RUnlock()
-
-	addresses = util.UniqueTrimmedStrings(addresses)
-	sort.Strings(addresses)
-
-	l.Debugln("lookup results for", deviceID)
-	l.Debugln("  addresses: ", addresses)
-
-	return addresses, nil
-}
-
-func (m *cachingMux) String() string {
-	return "discovery cache"
-}
-
-func (m *cachingMux) Error() error {
-	return nil
-}
-
-func (m *cachingMux) ChildErrors() map[string]error {
-	children := make(map[string]error, len(m.finders))
-	m.mut.RLock()
-	for _, f := range m.finders {
-		children[f.String()] = f.Error()
-	}
-	m.mut.RUnlock()
-	return children
-}
-
-func (m *cachingMux) Cache() map[protocol.DeviceID]CacheEntry {
-	// Res will be the "total" cache, i.e. the union of our cache and all our
-	// children's caches.
-	res := make(map[protocol.DeviceID]CacheEntry)
-
-	m.mut.RLock()
-	for i := range m.finders {
-		// Each finder[i] has a corresponding cache at cache[i]. Go through
-		// it and populate the total, appending any addresses and keeping
-		// the newest "when" time. We skip any negative cache entries.
-		for k, v := range m.caches[i].Cache() {
-			if v.found {
-				cur := res[k]
-				if v.when.After(cur.when) {
-					cur.when = v.when
-				}
-				cur.Addresses = append(cur.Addresses, v.Addresses...)
-				res[k] = cur
-			}
-		}
-
-		// Then ask the finder itself for its cache and do the same. If this
-		// finder is a global discovery client, it will have no cache. If it's
-		// a local discovery client, this will be its current state.
-		for k, v := range m.finders[i].Cache() {
-			if v.found {
-				cur := res[k]
-				if v.when.After(cur.when) {
-					cur.when = v.when
-				}
-				cur.Addresses = append(cur.Addresses, v.Addresses...)
-				res[k] = cur
-			}
-		}
-	}
-	m.mut.RUnlock()
-
-	for k, v := range res {
-		v.Addresses = util.UniqueTrimmedStrings(v.Addresses)
-		res[k] = v
-	}
-
-	return res
-}
-
 // A cache can be embedded wherever useful
 
 type cache struct {

lib/discover/cache_test.go

@@ -8,10 +8,13 @@ package discover
 
 import (
 	"context"
+	"crypto/tls"
 	"reflect"
 	"testing"
 	"time"
 
+	"github.com/syncthing/syncthing/lib/config"
+	"github.com/syncthing/syncthing/lib/events"
 	"github.com/syncthing/syncthing/lib/protocol"
 )
 
@@ -30,15 +33,19 @@ func TestCacheUnique(t *testing.T) {
 		"tcp://192.0.2.44:22000",
 	}
 
-	c := NewCachingMux()
-	c.(*cachingMux).ServeBackground()
+	cfg := config.New(protocol.LocalDeviceID)
+	cfg.Options.LocalAnnEnabled = false
+	cfg.Options.GlobalAnnEnabled = false
+
+	c := NewManager(protocol.LocalDeviceID, config.Wrap("", cfg, events.NoopLogger), tls.Certificate{}, events.NoopLogger, nil).(*manager)
+	c.ServeBackground()
 	defer c.Stop()
 
 	// Add a fake discovery service and verify we get its answers through the
 	// cache.
 
 	f1 := &fakeDiscovery{addresses0}
-	c.Add(f1, time.Minute, 0)
+	c.addLocked("f1", f1, time.Minute, 0)
 
 	ctx := context.Background()
 
@@ -54,7 +61,7 @@ func TestCacheUnique(t *testing.T) {
 	// duplicate or otherwise mess up the responses now.
 
 	f2 := &fakeDiscovery{addresses1}
-	c.Add(f2, time.Minute, 0)
+	c.addLocked("f2", f2, time.Minute, 0)
 
 	addr, err = c.Lookup(ctx, protocol.LocalDeviceID)
 	if err != nil {
@@ -86,15 +93,19 @@ func (f *fakeDiscovery) Cache() map[protocol.DeviceID]CacheEntry {
 }
 
 func TestCacheSlowLookup(t *testing.T) {
-	c := NewCachingMux()
-	c.(*cachingMux).ServeBackground()
+	cfg := config.New(protocol.LocalDeviceID)
+	cfg.Options.LocalAnnEnabled = false
+	cfg.Options.GlobalAnnEnabled = false
+
+	c := NewManager(protocol.LocalDeviceID, config.Wrap("", cfg, events.NoopLogger), tls.Certificate{}, events.NoopLogger, nil).(*manager)
+	c.ServeBackground()
 	defer c.Stop()
 
 	// Add a slow discovery service.
 
 	started := make(chan struct{})
 	f1 := &slowDiscovery{time.Second, started}
-	c.Add(f1, time.Minute, 0)
+	c.addLocked("f1", f1, time.Minute, 0)
 
 	// Start a lookup, which will take at least a second
 

lib/discover/discover.go

@@ -37,11 +37,6 @@ type FinderService interface {
 	suture.Service
 }
 
-type FinderMux interface {
-	Finder
-	ChildStatus() map[string]error
-}
-
 // The AddressLister answers questions about what addresses we are listening
 // on.
 type AddressLister interface {

lib/discover/global.go

@@ -12,6 +12,7 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -448,3 +449,15 @@ func (c *contextClient) Post(ctx context.Context, url, ctype string, data io.Rea
 	req.Header.Set("Content-Type", ctype)
 	return c.Client.Do(req)
 }
+
+func globalDiscoveryIdentity(addr string) string {
+	return "global discovery server " + addr
+}
+
+func ipv4Identity(port int) string {
+	return fmt.Sprintf("IPv4 local broadcast discovery on port %d", port)
+}
+
+func ipv6Identity(addr string) string {
+	return fmt.Sprintf("IPv6 local multicast discovery on address %s", addr)
+}

lib/discover/manager.go

@@ -0,0 +1,297 @@
+// Copyright (C) 2020 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at https://mozilla.org/MPL/2.0/.
+
+package discover
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/thejerf/suture"
+
+	"github.com/syncthing/syncthing/lib/config"
+	"github.com/syncthing/syncthing/lib/events"
+	"github.com/syncthing/syncthing/lib/protocol"
+	"github.com/syncthing/syncthing/lib/sync"
+	"github.com/syncthing/syncthing/lib/util"
+)
+
+// The Manager aggregates results from multiple Finders. Each Finder has
+// an associated cache time and negative cache time. The cache time sets how
+// long we cache and return successful lookup results, the negative cache
+// time sets how long we refrain from asking about the same device ID after
+// receiving a negative answer. The value of zero disables caching (positive
+// or negative).
+type Manager interface {
+	FinderService
+	ChildErrors() map[string]error
+}
+
+type manager struct {
+	*suture.Supervisor
+	myID          protocol.DeviceID
+	cfg           config.Wrapper
+	cert          tls.Certificate
+	evLogger      events.Logger
+	addressLister AddressLister
+
+	finders map[string]cachedFinder
+	mut     sync.RWMutex
+}
+
+func NewManager(myID protocol.DeviceID, cfg config.Wrapper, cert tls.Certificate, evLogger events.Logger, lister AddressLister) Manager {
+	return &manager{
+		Supervisor: suture.New("discover.Manager", suture.Spec{
+			PassThroughPanics: true,
+		}),
+		myID:          myID,
+		cfg:           cfg,
+		cert:          cert,
+		evLogger:      evLogger,
+		addressLister: lister,
+
+		finders: make(map[string]cachedFinder),
+		mut:     sync.NewRWMutex(),
+	}
+}
+
+func (m *manager) Serve() {
+	m.cfg.Subscribe(m)
+	defer m.cfg.Unsubscribe(m)
+	m.CommitConfiguration(config.Configuration{}, m.cfg.RawCopy())
+	m.Supervisor.Serve()
+}
+
+func (m *manager) addLocked(identity string, finder Finder, cacheTime, negCacheTime time.Duration) {
+	entry := cachedFinder{
+		Finder:       finder,
+		cacheTime:    cacheTime,
+		negCacheTime: negCacheTime,
+		cache:        newCache(),
+		token:        nil,
+	}
+	if service, ok := finder.(suture.Service); ok {
+		token := m.Supervisor.Add(service)
+		entry.token = &token
+	}
+	m.finders[identity] = entry
+	l.Infoln("Using discovery mechanism:", identity)
+}
+
+func (m *manager) removeLocked(identity string) {
+	entry, ok := m.finders[identity]
+	if !ok {
+		return
+	}
+	if entry.token != nil {
+		err := m.Supervisor.Remove(*entry.token)
+		if err != nil {
+			l.Warnf("removing discovery %s: %s", identity, err)
+		}
+	}
+	delete(m.finders, identity)
+	l.Infoln("Stopped using discovery mechanism: ", identity)
+}
+
+// Lookup attempts to resolve the device ID using any of the added Finders,
+// while obeying the cache settings.
+func (m *manager) Lookup(ctx context.Context, deviceID protocol.DeviceID) (addresses []string, err error) {
+	m.mut.RLock()
+	for _, finder := range m.finders {
+		if cacheEntry, ok := finder.cache.Get(deviceID); ok {
+			// We have a cache entry. Lets see what it says.
+
+			if cacheEntry.found && time.Since(cacheEntry.when) < finder.cacheTime {
+				// It's a positive, valid entry. Use it.
+				l.Debugln("cached discovery entry for", deviceID, "at", finder)
+				l.Debugln("  cache:", cacheEntry)
+				addresses = append(addresses, cacheEntry.Addresses...)
+				continue
+			}
+
+			valid := time.Now().Before(cacheEntry.validUntil) || time.Since(cacheEntry.when) < finder.negCacheTime
+			if !cacheEntry.found && valid {
+				// It's a negative, valid entry. We should not make another
+				// attempt right now.
+				l.Debugln("negative cache entry for", deviceID, "at", finder, "valid until", cacheEntry.when.Add(finder.negCacheTime), "or", cacheEntry.validUntil)
+				continue
+			}
+
+			// It's expired. Ignore and continue.
+		}
+
+		// Perform the actual lookup and cache the result.
+		if addrs, err := finder.Lookup(ctx, deviceID); err == nil {
+			l.Debugln("lookup for", deviceID, "at", finder)
+			l.Debugln("  addresses:", addrs)
+			addresses = append(addresses, addrs...)
+			finder.cache.Set(deviceID, CacheEntry{
+				Addresses: addrs,
+				when:      time.Now(),
+				found:     len(addrs) > 0,
+			})
+		} else {
+			// Lookup returned error, add a negative cache entry.
+			entry := CacheEntry{
+				when:  time.Now(),
+				found: false,
+			}
+			if err, ok := err.(cachedError); ok {
+				entry.validUntil = time.Now().Add(err.CacheFor())
+			}
+			finder.cache.Set(deviceID, entry)
+		}
+	}
+	m.mut.RUnlock()
+
+	addresses = util.UniqueTrimmedStrings(addresses)
+	sort.Strings(addresses)
+
+	l.Debugln("lookup results for", deviceID)
+	l.Debugln("  addresses: ", addresses)
+
+	return addresses, nil
+}
+
+func (m *manager) String() string {
+	return "discovery cache"
+}
+
+func (m *manager) Error() error {
+	return nil
+}
+
+func (m *manager) ChildErrors() map[string]error {
+	children := make(map[string]error, len(m.finders))
+	m.mut.RLock()
+	for _, f := range m.finders {
+		children[f.String()] = f.Error()
+	}
+	m.mut.RUnlock()
+	return children
+}
+
+func (m *manager) Cache() map[protocol.DeviceID]CacheEntry {
+	// Res will be the "total" cache, i.e. the union of our cache and all our
+	// children's caches.
+	res := make(map[protocol.DeviceID]CacheEntry)
+
+	m.mut.RLock()
+	for _, finder := range m.finders {
+		// Each finder[i] has a corresponding cache. Go through
+		// it and populate the total, appending any addresses and keeping
+		// the newest "when" time. We skip any negative cache finders.
+		for k, v := range finder.cache.Cache() {
+			if v.found {
+				cur := res[k]
+				if v.when.After(cur.when) {
+					cur.when = v.when
+				}
+				cur.Addresses = append(cur.Addresses, v.Addresses...)
+				res[k] = cur
+			}
+		}
+
+		// Then ask the finder itself for its cache and do the same. If this
+		// finder is a global discovery client, it will have no cache. If it's
+		// a local discovery client, this will be its current state.
+		for k, v := range finder.Cache() {
+			if v.found {
+				cur := res[k]
+				if v.when.After(cur.when) {
+					cur.when = v.when
+				}
+				cur.Addresses = append(cur.Addresses, v.Addresses...)
+				res[k] = cur
+			}
+		}
+	}
+	m.mut.RUnlock()
+
+	for k, v := range res {
+		v.Addresses = util.UniqueTrimmedStrings(v.Addresses)
+		res[k] = v
+	}
+
+	return res
+}
+
+func (m *manager) VerifyConfiguration(_, _ config.Configuration) error {
+	return nil
+}
+
+func (m *manager) CommitConfiguration(_, to config.Configuration) (handled bool) {
+	m.mut.Lock()
+	defer m.mut.Unlock()
+	toIdentities := make(map[string]struct{})
+	if to.Options.GlobalAnnEnabled {
+		for _, srv := range to.Options.GlobalDiscoveryServers() {
+			toIdentities[globalDiscoveryIdentity(srv)] = struct{}{}
+		}
+	}
+
+	if to.Options.LocalAnnEnabled {
+		toIdentities[ipv4Identity(to.Options.LocalAnnPort)] = struct{}{}
+		toIdentities[ipv6Identity(to.Options.LocalAnnMCAddr)] = struct{}{}
+	}
+
+	// Remove things that we're not expected to have.
+	for identity := range m.finders {
+		if _, ok := toIdentities[identity]; !ok {
+			m.removeLocked(identity)
+		}
+	}
+
+	// Add things we don't have.
+	if to.Options.GlobalAnnEnabled {
+		for _, srv := range to.Options.GlobalDiscoveryServers() {
+			identity := globalDiscoveryIdentity(srv)
+			// Skip, if it's already running.
+			if _, ok := m.finders[identity]; ok {
+				continue
+			}
+			gd, err := NewGlobal(srv, m.cert, m.addressLister, m.evLogger)
+			if err != nil {
+				l.Warnln("Global discovery:", err)
+				continue
+			}
+
+			// Each global discovery server gets its results cached for five
+			// minutes, and is not asked again for a minute when it's returned
+			// unsuccessfully.
+			m.addLocked(identity, gd, 5*time.Minute, time.Minute)
+		}
+	}
+
+	if to.Options.LocalAnnEnabled {
+		// v4 broadcasts
+		v4Identity := ipv4Identity(to.Options.LocalAnnPort)
+		if _, ok := m.finders[v4Identity]; !ok {
+			bcd, err := NewLocal(m.myID, fmt.Sprintf(":%d", to.Options.LocalAnnPort), m.addressLister, m.evLogger)
+			if err != nil {
+				l.Warnln("IPv4 local discovery:", err)
+			} else {
+				m.addLocked(v4Identity, bcd, 0, 0)
+			}
+		}
+
+		// v6 multicasts
+		v6Identity := ipv6Identity(to.Options.LocalAnnMCAddr)
+		if _, ok := m.finders[v6Identity]; !ok {
+			mcd, err := NewLocal(m.myID, to.Options.LocalAnnMCAddr, m.addressLister, m.evLogger)
+			if err != nil {
+				l.Warnln("IPv6 local discovery:", err)
+			} else {
+				m.addLocked(v6Identity, mcd, 0, 0)
+			}
+		}
+	}
+
+	return true
+}