felixfoertsch/syncthing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

all: Store pending devices and folders in database (fixes #7178) (#6443)

Browse Source
This commit is contained in:
André Colomb
2020-12-17 19:54:31 +01:00
committed by GitHub
parent 4470cd5aaa
commit 7502997e7e
20 changed files with 1259 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/model/model.go
+96 -6
View File
@@ -105,6 +105,9 @@ type Model interface {
FolderStatistics() (map[string]stats.FolderStatistics, error)
UsageReportingStats(report *contract.Report, version int, preview bool)
PendingDevices() (map[protocol.DeviceID]db.ObservedDevice, error)
PendingFolders(device protocol.DeviceID) (map[string]db.PendingFolder, error)
StartDeadlockDetector(timeout time.Duration)
GlobalDirectoryTree(folder, prefix string, levels int, dirsonly bool) map[string]interface{}
}
@@ -255,8 +258,10 @@ func NewModel(cfg config.Wrapper, id protocol.DeviceID, clientName, clientVersio
func (m *model) serve(ctx context.Context) error {
// Add and start folders
cacheIgnoredFiles := m.cfg.Options().CacheIgnoredFiles
clusterConfigDevices := make(deviceIDSet, len(m.cfg.Devices()))
for _, folderCfg := range m.cfg.Folders() {
existingDevices := m.cfg.Devices()
existingFolders := m.cfg.Folders()
clusterConfigDevices := make(deviceIDSet, len(existingDevices))
for _, folderCfg := range existingFolders {
if folderCfg.Paused {
folderCfg.CreateRoot()
continue
@@ -264,6 +269,10 @@ func (m *model) serve(ctx context.Context) error {
m.newFolder(folderCfg, cacheIgnoredFiles)
clusterConfigDevices.add(folderCfg.DeviceIDs())
}
ignoredDevices := observedDeviceSet(m.cfg.IgnoredDevices())
m.cleanPending(existingDevices, existingFolders, ignoredDevices, nil)
m.resendClusterConfig(clusterConfigDevices.AsSlice())
m.cfg.Subscribe(m)
@@ -1251,9 +1260,10 @@ func (m *model) ccHandleFolders(folders []protocol.Folder, deviceCfg config.Devi
l.Infof("Ignoring folder %s from device %s since we are configured to", folder.Description(), deviceID)
continue
}
m.cfg.AddOrUpdatePendingFolder(folder.ID, folder.Label, deviceID)
if err := m.db.AddOrUpdatePendingFolder(folder.ID, folder.Label, deviceID); err != nil {
l.Warnf("Failed to persist pending folder entry to database: %v", err)
}
indexSenders.addPending(cfg, ccDeviceInfos[folder.ID])
changed = true
m.evLogger.Log(events.FolderRejected, map[string]string{
"folder": folder.ID,
"folderLabel": folder.Label,
@@ -1989,8 +1999,9 @@ func (m *model) OnHello(remoteID protocol.DeviceID, addr net.Addr, hello protoco
cfg, ok := m.cfg.Device(remoteID)
if !ok {
m.cfg.AddOrUpdatePendingDevice(remoteID, hello.DeviceName, addr.String())
_ = m.cfg.Save() // best effort
if err := m.db.AddOrUpdatePendingDevice(remoteID, hello.DeviceName, addr.String()); err != nil {
l.Warnf("Failed to persist pending device entry to database: %v", err)
}
m.evLogger.Log(events.DeviceRejected, map[string]string{
"name": hello.DeviceName,
"device": remoteID.String(),
@@ -2577,12 +2588,14 @@ func (m *model) CommitConfiguration(from, to config.Configuration) bool {
}
}
removedFolders := make(map[string]struct{})
for folderID, fromCfg := range fromFolders {
toCfg, ok := toFolders[folderID]
if !ok {
// The folder was removed.
m.removeFolder(fromCfg)
clusterConfigDevices.add(fromCfg.DeviceIDs())
removedFolders[fromCfg.ID] = struct{}{}
continue
}
@@ -2647,6 +2660,7 @@ func (m *model) CommitConfiguration(from, to config.Configuration) bool {
m.evLogger.Log(events.DeviceResumed, map[string]string{"device": deviceID.String()})
}
}
// Clean up after removed devices
removedDevices := make([]protocol.DeviceID, 0, len(fromDevices))
m.fmut.Lock()
for deviceID := range fromDevices {
@@ -2671,6 +2685,9 @@ func (m *model) CommitConfiguration(from, to config.Configuration) bool {
// Generating cluster-configs acquires fmut -> must happen outside of pmut.
m.resendClusterConfig(clusterConfigDevices.AsSlice())
ignoredDevices := observedDeviceSet(to.IgnoredDevices)
m.cleanPending(toDevices, toFolders, ignoredDevices, removedFolders)
m.globalRequestLimiter.setCapacity(1024 * to.Options.MaxConcurrentIncomingRequestKiB())
m.folderIOLimiter.setCapacity(to.Options.MaxFolderConcurrency())
@@ -2685,6 +2702,59 @@ func (m *model) CommitConfiguration(from, to config.Configuration) bool {
return true
}
func (m *model) cleanPending(existingDevices map[protocol.DeviceID]config.DeviceConfiguration, existingFolders map[string]config.FolderConfiguration, ignoredDevices deviceIDSet, removedFolders map[string]struct{}) {
pendingFolders, err := m.db.PendingFolders()
if err != nil {
l.Infof("Could not iterate through pending folder entries for cleanup: %v", err)
// Continue with pending devices below, loop is skipped.
}
for folderID, pf := range pendingFolders {
if _, ok := removedFolders[folderID]; ok {
// Forget pending folder device associations for recently removed
// folders as well, assuming the folder is no longer of interest
// at all (but might become pending again).
l.Debugf("Discarding pending removed folder %v from all devices", folderID)
m.db.RemovePendingFolder(folderID)
continue
}
for deviceID := range pf.OfferedBy {
if dev, ok := existingDevices[deviceID]; !ok {
l.Debugf("Discarding pending folder %v from unknown device %v", folderID, deviceID)
m.db.RemovePendingFolderForDevice(folderID, deviceID)
continue
} else if dev.IgnoredFolder(folderID) {
l.Debugf("Discarding now ignored pending folder %v for device %v", folderID, deviceID)
m.db.RemovePendingFolderForDevice(folderID, deviceID)
continue
}
if folderCfg, ok := existingFolders[folderID]; ok {
if folderCfg.SharedWith(deviceID) {
l.Debugf("Discarding now shared pending folder %v for device %v", folderID, deviceID)
m.db.RemovePendingFolderForDevice(folderID, deviceID)
}
}
}
}
pendingDevices, err := m.db.PendingDevices()
if err != nil {
l.Infof("Could not iterate through pending device entries for cleanup: %v", err)
return
}
for deviceID := range pendingDevices {
if _, ok := ignoredDevices[deviceID]; ok {
l.Debugf("Discarding now ignored pending device %v", deviceID)
m.db.RemovePendingDevice(deviceID)
continue
}
if _, ok := existingDevices[deviceID]; ok {
l.Debugf("Discarding now added pending device %v", deviceID)
m.db.RemovePendingDevice(deviceID)
continue
}
}
}
// checkFolderRunningLocked returns nil if the folder is up and running and a
// descriptive error if not.
// Need to hold (read) lock on m.fmut when calling this.
@@ -2703,6 +2773,18 @@ func (m *model) checkFolderRunningLocked(folder string) error {
return errFolderNotRunning
}
// PendingDevices lists unknown devices that tried to connect.
func (m *model) PendingDevices() (map[protocol.DeviceID]db.ObservedDevice, error) {
return m.db.PendingDevices()
}
// PendingFolders lists folders that we don't yet share with the offering devices. It
// returns the entries grouped by folder and filters for a given device unless the
// argument is specified as EmptyDeviceID.
func (m *model) PendingFolders(device protocol.DeviceID) (map[string]db.PendingFolder, error) {
return m.db.PendingFoldersForDevice(device)
}
// mapFolders returns a map of folder ID to folder configuration for the given
// slice of folder configurations.
func mapFolders(folders []config.FolderConfiguration) map[string]config.FolderConfiguration {
@@ -2723,6 +2805,14 @@ func mapDevices(devices []protocol.DeviceID) map[protocol.DeviceID]struct{} {
return m
}
func observedDeviceSet(devices []config.ObservedDevice) deviceIDSet {
res := make(deviceIDSet, len(devices))
for _, dev := range devices {
res[dev.ID] = struct{}{}
}
return res
}
func readOffsetIntoBuf(fs fs.Filesystem, file string, offset int64, buf []byte) error {
fd, err := fs.Open(file)
if err != nil {
lib/model/model_test.go
+72
View File
@@ -4323,6 +4323,78 @@ func TestCCFolderNotRunning(t *testing.T) {
}
}
func TestPendingFolder(t *testing.T) {
w, _ := tmpDefaultWrapper()
m := setupModel(w)
defer cleanupModel(m)
waiter, err := w.SetDevice(config.DeviceConfiguration{DeviceID: device2})
if err != nil {
t.Fatal(err)
}
waiter.Wait()
pfolder := "default"
if err := m.db.AddOrUpdatePendingFolder(pfolder, pfolder, device2); err != nil {
t.Fatal(err)
}
deviceFolders, err := m.PendingFolders(protocol.EmptyDeviceID)
if err != nil {
t.Fatal(err)
} else if pf, ok := deviceFolders[pfolder]; !ok {
t.Errorf("folder %v not pending", pfolder)
} else if _, ok := pf.OfferedBy[device2]; !ok {
t.Errorf("folder %v not pending for device %v", pfolder, device2)
} else if len(pf.OfferedBy) > 1 {
t.Errorf("folder %v pending for too many devices %v", pfolder, pf.OfferedBy)
}
device3, err := protocol.DeviceIDFromString("AIBAEAQ-CAIBAEC-AQCAIBA-EAQCAIA-BAEAQCA-IBAEAQC-CAIBAEA-QCAIBA7")
waiter, err = w.SetDevice(config.DeviceConfiguration{DeviceID: device3})
if err != nil {
t.Fatal(err)
}
waiter.Wait()
if err := m.db.AddOrUpdatePendingFolder(pfolder, pfolder, device3); err != nil {
t.Fatal(err)
}
deviceFolders, err = m.PendingFolders(device2)
if err != nil {
t.Fatal(err)
} else if pf, ok := deviceFolders[pfolder]; !ok {
t.Errorf("folder %v not pending when filtered", pfolder)
} else if _, ok := pf.OfferedBy[device2]; !ok {
t.Errorf("folder %v not pending for device %v when filtered", pfolder, device2)
} else if _, ok := pf.OfferedBy[device3]; ok {
t.Errorf("folder %v pending for device %v, but not filtered out", pfolder, device3)
}
waiter, err = w.RemoveDevice(device3)
if err != nil {
t.Fatal(err)
}
waiter.Wait()
deviceFolders, err = m.PendingFolders(protocol.EmptyDeviceID)
if err != nil {
t.Fatal(err)
} else if pf, ok := deviceFolders[pfolder]; !ok {
t.Errorf("folder %v not pending", pfolder)
} else if _, ok := pf.OfferedBy[device3]; ok {
t.Errorf("folder %v pending for removed device %v", pfolder, device3)
}
waiter, err = w.RemoveFolder(pfolder)
if err != nil {
t.Fatal(err)
}
waiter.Wait()
deviceFolders, err = m.PendingFolders(protocol.EmptyDeviceID)
if err != nil {
t.Fatal(err)
} else if _, ok := deviceFolders[pfolder]; ok {
t.Errorf("folder %v still pending after local removal", pfolder)
}
}
func equalStringsInAnyOrder(a, b []string) bool {
if len(a) != len(b) {
return false