lib/config, lib/connections: Add optional connection limits (fixes #7176) (#7223)

This adds two new configuration options:

    // The number of connections at which we stop trying to connect to more
    // devices, zero meaning no limit. Does not affect incoming connections.
    ConnectionLimitEnough int

    // The maximum number of connections which we will allow in total, zero
    // meaning no limit. Affects incoming connections and prevents
    // attempting outgoing connections.
    ConnectionLimitMax int

These can be used to limit the number of concurrent connections in
various ways.

lib/connections/dialqueue.go

@@ -0,0 +1,54 @@
+// Copyright (C) 2021 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 connections
+
+import (
+	"sort"
+	"time"
+
+	"github.com/syncthing/syncthing/lib/protocol"
+	"github.com/syncthing/syncthing/lib/rand"
+)
+
+type dialQueueEntry struct {
+	id         protocol.DeviceID
+	lastSeen   time.Time
+	shortLived bool
+	targets    []dialTarget
+}
+
+type dialQueue []dialQueueEntry
+
+func (queue dialQueue) Sort() {
+	// Sort the queue with the most recently seen device at the head,
+	// increasing the likelihood of connecting to a device that we're
+	// already almost up to date with, index wise.
+	sort.Slice(queue, func(a, b int) bool {
+		qa, qb := queue[a], queue[b]
+		if qa.shortLived != qb.shortLived {
+			return qb.shortLived
+		}
+		return qa.lastSeen.After(qb.lastSeen)
+	})
+
+	// Shuffle the part of the connection queue that are devices we haven't
+	// connected to recently, so that if we only try a limited set of
+	// devices (or they in turn have limits and we're trying to load balance
+	// over several) and the usual ones are down it won't be the same ones
+	// in the same order every time.
+	idx := 0
+	cutoff := time.Now().Add(-recentlySeenCutoff)
+	for idx < len(queue) {
+		if queue[idx].lastSeen.Before(cutoff) {
+			break
+		}
+		idx++
+	}
+	if idx < len(queue)-1 {
+		rand.Shuffle(queue[idx:])
+	}
+}

lib/connections/dialqueue_test.go

@@ -0,0 +1,119 @@
+// Copyright (C) 2021 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 connections
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/syncthing/syncthing/lib/protocol"
+)
+
+func TestDialQueueSort(t *testing.T) {
+	t.Parallel()
+
+	t.Run("ByLastSeen", func(t *testing.T) {
+		t.Parallel()
+
+		// Devices seen within the last week or so should be sorted stricly in order.
+		now := time.Now()
+		queue := dialQueue{
+			{id: device1, lastSeen: now.Add(-5 * time.Hour)},  // 1
+			{id: device2, lastSeen: now.Add(-50 * time.Hour)}, // 3
+			{id: device3, lastSeen: now.Add(-25 * time.Hour)}, // 2
+			{id: device4, lastSeen: now.Add(-2 * time.Hour)},  // 0
+		}
+		expected := []protocol.ShortID{device4.Short(), device1.Short(), device3.Short(), device2.Short()}
+
+		queue.Sort()
+
+		if !reflect.DeepEqual(shortDevices(queue), expected) {
+			t.Error("expected different order")
+		}
+	})
+
+	t.Run("OldConnections", func(t *testing.T) {
+		t.Parallel()
+
+		// Devices seen long ago should be randomized.
+		now := time.Now()
+		queue := dialQueue{
+			{id: device1, lastSeen: now.Add(-5 * time.Hour)},       // 1
+			{id: device2, lastSeen: now.Add(-50 * 24 * time.Hour)}, // 2, 3
+			{id: device3, lastSeen: now.Add(-25 * 24 * time.Hour)}, // 2, 3
+			{id: device4, lastSeen: now.Add(-2 * time.Hour)},       // 0
+		}
+
+		expected1 := []protocol.ShortID{device4.Short(), device1.Short(), device3.Short(), device2.Short()}
+		expected2 := []protocol.ShortID{device4.Short(), device1.Short(), device2.Short(), device3.Short()}
+
+		var seen1, seen2 int
+
+		for i := 0; i < 100; i++ {
+			queue.Sort()
+			res := shortDevices(queue)
+			if reflect.DeepEqual(res, expected1) {
+				seen1++
+				continue
+			}
+			if reflect.DeepEqual(res, expected2) {
+				seen2++
+				continue
+			}
+			t.Fatal("expected different order")
+		}
+
+		if seen1 < 10 || seen2 < 10 {
+			t.Error("expected more even distribution", seen1, seen2)
+		}
+	})
+
+	t.Run("ShortLivedConnections", func(t *testing.T) {
+		t.Parallel()
+
+		// Short lived connections should be sorted as if they were long ago
+		now := time.Now()
+		queue := dialQueue{
+			{id: device1, lastSeen: now.Add(-5 * time.Hour)},                   // 1
+			{id: device2, lastSeen: now.Add(-3 * time.Hour)},                   // 0
+			{id: device3, lastSeen: now.Add(-25 * 24 * time.Hour)},             // 2, 3
+			{id: device4, lastSeen: now.Add(-2 * time.Hour), shortLived: true}, // 2, 3
+		}
+
+		expected1 := []protocol.ShortID{device2.Short(), device1.Short(), device3.Short(), device4.Short()}
+		expected2 := []protocol.ShortID{device2.Short(), device1.Short(), device4.Short(), device3.Short()}
+
+		var seen1, seen2 int
+
+		for i := 0; i < 100; i++ {
+			queue.Sort()
+			res := shortDevices(queue)
+			if reflect.DeepEqual(res, expected1) {
+				seen1++
+				continue
+			}
+			if reflect.DeepEqual(res, expected2) {
+				seen2++
+				continue
+			}
+			t.Fatal("expected different order")
+		}
+
+		if seen1 < 10 || seen2 < 10 {
+			t.Error("expected more even distribution", seen1, seen2)
+		}
+	})
+}
+
+func shortDevices(queue dialQueue) []protocol.ShortID {
+	res := make([]protocol.ShortID, len(queue))
+	for i, qe := range queue {
+		res[i] = qe.id.Short()
+	}
+	return res
+}

lib/connections/service.go

@@ -55,12 +55,14 @@ var (
 )
 
 const (
-	perDeviceWarningIntv    = 15 * time.Minute
-	tlsHandshakeTimeout     = 10 * time.Second
-	minConnectionReplaceAge = 10 * time.Second
-	minConnectionLoopSleep  = 5 * time.Second
-	stdConnectionLoopSleep  = time.Minute
-	worstDialerPriority     = math.MaxInt32
+	perDeviceWarningIntv          = 15 * time.Minute
+	tlsHandshakeTimeout           = 10 * time.Second
+	minConnectionReplaceAge       = 10 * time.Second
+	minConnectionLoopSleep        = 5 * time.Second
+	stdConnectionLoopSleep        = time.Minute
+	worstDialerPriority           = math.MaxInt32
+	recentlySeenCutoff            = 7 * 24 * time.Hour
+	shortLivedConnectionThreshold = 5 * time.Second
 )
 
 // From go/src/crypto/tls/cipher_suites.go
@@ -415,6 +417,24 @@ func (s *service) bestDialerPriority(cfg config.Configuration) int {
 }
 
 func (s *service) dialDevices(ctx context.Context, now time.Time, cfg config.Configuration, bestDialerPriority int, nextDialAt map[string]time.Time, initial bool) {
+	// Figure out current connection limits up front to see if there's any
+	// point in resolving devices and such at all.
+	allowAdditional := 0 // no limit
+	connectionLimit := cfg.Options.LowestConnectionLimit()
+	if connectionLimit > 0 {
+		current := s.model.NumConnections()
+		allowAdditional = connectionLimit - current
+		if allowAdditional <= 0 {
+			l.Debugf("Skipping dial because we've reached the connection limit, current %d >= limit %d", current, connectionLimit)
+			return
+		}
+	}
+
+	// Get device statistics for the last seen time of each device. This
+	// isn't critical, so ignore the potential error.
+	stats, _ := s.model.DeviceStatistics()
+
+	queue := make(dialQueue, 0, len(cfg.Devices))
 	for _, deviceCfg := range cfg.Devices {
 		// Don't attempt to connect to ourselves...
 		if deviceCfg.DeviceID == s.myID {
@@ -440,8 +460,34 @@ func (s *service) dialDevices(ctx context.Context, now time.Time, cfg config.Con
 		}
 
 		dialTargets := s.resolveDialTargets(ctx, now, cfg, deviceCfg, nextDialAt, initial, priorityCutoff)
-		if conn, ok := s.dialParallel(ctx, deviceCfg.DeviceID, dialTargets); ok {
+		if len(dialTargets) > 0 {
+			queue = append(queue, dialQueueEntry{
+				id:         deviceCfg.DeviceID,
+				lastSeen:   stats[deviceCfg.DeviceID].LastSeen,
+				shortLived: stats[deviceCfg.DeviceID].LastConnectionDurationS < shortLivedConnectionThreshold.Seconds(),
+				targets:    dialTargets,
+			})
+		}
+	}
+
+	// Sort the queue in an order we think will be useful (most recent
+	// first, deprioriting unstable devices, randomizing those we haven't
+	// seen in a long while). If we don't do connection limiting the sorting
+	// doesn't have much effect, but it may result in getting up and running
+	// quicker if only a subset of configured devices are actually reachable
+	// (by prioritizing those that were reachable recently).
+	dialQueue.Sort(queue)
+
+	// Perform dials according to the queue, stopping when we've reached the
+	// allowed additional number of connections (if limited).
+	numConns := 0
+	for _, entry := range queue {
+		if conn, ok := s.dialParallel(ctx, entry.id, entry.targets); ok {
 			s.conns <- conn
+			numConns++
+			if allowAdditional > 0 && numConns >= allowAdditional {
+				break
+			}
 		}
 	}
 }

lib/connections/structs.go

@@ -18,6 +18,7 @@ import (
 	"github.com/syncthing/syncthing/lib/config"
 	"github.com/syncthing/syncthing/lib/nat"
 	"github.com/syncthing/syncthing/lib/protocol"
+	"github.com/syncthing/syncthing/lib/stats"
 
 	"github.com/thejerf/suture/v4"
 )
@@ -193,9 +194,11 @@ type genericListener interface {
 type Model interface {
 	protocol.Model
 	AddConnection(conn protocol.Connection, hello protocol.Hello)
+	NumConnections() int
 	Connection(remoteID protocol.DeviceID) (protocol.Connection, bool)
 	OnHello(protocol.DeviceID, net.Addr, protocol.Hello) error
 	GetHello(protocol.DeviceID) protocol.HelloIntf
+	DeviceStatistics() (map[protocol.DeviceID]stats.DeviceStatistics, error)
 }
 
 type onAddressesChangedNotifier struct {