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

chore(model): refactor context handling for folder type (#10472)

Browse Source 
Signed-off-by: Jakob Borg <jakob@kastelo.net>
This commit is contained in:
Jakob Borg
2025-11-27 20:34:35 +00:00
committed by GitHub
parent ce884e5d72
commit 553c02f244
12 changed files with 339 additions and 404 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/model/folder.go
+89 -90
View File
@@ -55,8 +55,7 @@ type folder struct {
ignores *ignore.Matcher
mtimefs fs.Filesystem
modTimeWindow time.Duration
ctx context.Context //nolint:containedctx // used internally, only accessible on serve lifetime
done chan struct{} // used externally, accessible regardless of serve
done chan struct{} // used externally, accessible regardless of serve
sl *slog.Logger
scanInterval time.Duration
@@ -94,12 +93,12 @@ type folder struct {
}
type syncRequest struct {
fn func() error
fn func(context.Context) error
err chan error
}
type puller interface {
pull() (bool, error) // true when successful and should not be retried
pull(ctx context.Context) (bool, error) // true when successful and should not be retried
}
func newFolder(model *model, ignores *ignore.Matcher, cfg config.FolderConfiguration, evLogger events.Logger, ioLimiter *semaphore.Semaphore, ver versioner.Versioner) *folder {
@@ -151,10 +150,8 @@ func (f *folder) Serve(ctx context.Context) error {
f.model.foldersRunning.Add(1)
defer f.model.foldersRunning.Add(-1)
f.ctx = ctx
l.Debugln(f, "starting")
defer l.Debugln(f, "exiting")
f.sl.DebugContext(ctx, "Folder starting")
defer f.sl.DebugContext(ctx, "Folder exiting")
defer func() {
f.scanTimer.Stop()
@@ -163,7 +160,7 @@ func (f *folder) Serve(ctx context.Context) error {
}()
if f.FSWatcherEnabled && f.getHealthErrorAndLoadIgnores() == nil {
f.startWatch()
f.startWatch(ctx)
}
// If we're configured to not do version cleanup, or we don't have a
@@ -182,7 +179,7 @@ func (f *folder) Serve(ctx context.Context) error {
var err error
select {
case <-f.ctx.Done():
case <-ctx.Done():
close(f.done)
return nil
@@ -197,16 +194,16 @@ func (f *folder) Serve(ctx context.Context) error {
}
pullTimer.Reset(time.Duration(float64(time.Second) * f.PullerDelayS))
} else {
_, err = f.pull()
_, err = f.pull(ctx)
}
case <-pullTimer.C:
f.setState(FolderIdle)
_, err = f.pull()
_, err = f.pull(ctx)
case <-f.pullFailTimer.C:
var success bool
success, err = f.pull()
success, err = f.pull(ctx)
if (err != nil || !success) && f.pullPause < 60*f.pullBasePause() {
// Back off from retrying to pull
f.pullPause *= 2
@@ -215,46 +212,46 @@ func (f *folder) Serve(ctx context.Context) error {
case <-initialCompleted:
// Initial scan has completed, we should do a pull
initialCompleted = nil // never hit this case again
_, err = f.pull()
_, err = f.pull(ctx)
case <-f.forcedRescanRequested:
err = f.handleForcedRescans()
err = f.handleForcedRescans(ctx)
case <-f.scanTimer.C:
l.Debugln(f, "Scanning due to timer")
err = f.scanTimerFired()
f.sl.DebugContext(ctx, "Scanning due to timer")
err = f.scanTimerFired(ctx)
case req := <-f.doInSyncChan:
l.Debugln(f, "Running something due to request")
err = req.fn()
f.sl.DebugContext(ctx, "Running something due to request")
err = req.fn(ctx)
req.err <- err
case next := <-f.scanDelay:
l.Debugln(f, "Delaying scan")
f.sl.DebugContext(ctx, "Delaying scan")
f.scanTimer.Reset(next)
case <-f.scanScheduled:
l.Debugln(f, "Scan was scheduled")
f.sl.DebugContext(ctx, "Scan was scheduled")
f.scanTimer.Reset(0)
case fsEvents := <-f.watchChan:
l.Debugln(f, "Scan due to watcher")
err = f.scanSubdirs(fsEvents)
f.sl.DebugContext(ctx, "Scan due to watcher")
err = f.scanSubdirs(ctx, fsEvents)
case <-f.restartWatchChan:
l.Debugln(f, "Restart watcher")
err = f.restartWatch()
f.sl.DebugContext(ctx, "Restart watcher")
err = f.restartWatch(ctx)
case <-f.versionCleanupTimer.C:
l.Debugln(f, "Doing version cleanup")
f.versionCleanupTimerFired()
f.sl.DebugContext(ctx, "Doing version cleanup")
f.versionCleanupTimerFired(ctx)
}
if err != nil {
if svcutil.IsFatal(err) {
return err
}
f.setError(err)
f.setError(ctx, err)
}
}
}
@@ -303,12 +300,14 @@ func (*folder) Jobs(_, _ int) ([]string, []string, int) {
func (f *folder) Scan(subdirs []string) error {
<-f.initialScanFinished
return f.doInSync(func() error { return f.scanSubdirs(subdirs) })
return f.doInSync(func(ctx context.Context) error {
return f.scanSubdirs(ctx, subdirs)
})
}
// doInSync allows to run functions synchronously in folder.serve from exported,
// asynchronously called methods.
func (f *folder) doInSync(fn func() error) error {
func (f *folder) doInSync(fn func(context.Context) error) error {
req := syncRequest{
fn: fn,
err: make(chan error, 1),
@@ -329,7 +328,7 @@ func (f *folder) Reschedule() {
// Sleep a random time between 3/4 and 5/4 of the configured interval.
sleepNanos := (f.scanInterval.Nanoseconds()*3 + rand.Int63n(2*f.scanInterval.Nanoseconds())) / 4 //nolint:gosec
interval := time.Duration(sleepNanos) * time.Nanosecond
l.Debugln(f, "next rescan in", interval)
f.sl.Debug("Next rescan scheduled", slog.Duration("interval", interval))
f.scanTimer.Reset(interval)
}
@@ -365,7 +364,7 @@ func (f *folder) getHealthErrorWithoutIgnores() error {
return nil
}
func (f *folder) pull() (success bool, err error) {
func (f *folder) pull(ctx context.Context) (success bool, err error) {
f.pullFailTimer.Stop()
select {
case <-f.pullFailTimer.C:
@@ -402,7 +401,7 @@ func (f *folder) pull() (success bool, err error) {
// Abort early (before acquiring a token) if there's a folder error
err = f.getHealthErrorWithoutIgnores()
if err != nil {
l.Debugln("Skipping pull of", f.Description(), "due to folder error:", err)
f.sl.DebugContext(ctx, "Skipping pull due to folder error", slogutil.Error(err))
return false, err
}
@@ -411,7 +410,7 @@ func (f *folder) pull() (success bool, err error) {
if f.Type != config.FolderTypeSendOnly {
f.setState(FolderSyncWaiting)
if err := f.ioLimiter.TakeWithContext(f.ctx, 1); err != nil {
if err := f.ioLimiter.TakeWithContext(ctx, 1); err != nil {
return true, err
}
defer f.ioLimiter.Give(1)
@@ -428,12 +427,12 @@ func (f *folder) pull() (success bool, err error) {
}()
err = f.getHealthErrorAndLoadIgnores()
if err != nil {
l.Debugln("Skipping pull of", f.Description(), "due to folder error:", err)
f.sl.DebugContext(ctx, "Skipping pull due to folder error", slogutil.Error(err))
return false, err
}
f.setError(nil)
f.setError(ctx, nil)
success, err = f.puller.pull()
success, err = f.puller.pull(ctx)
if success && err == nil {
return true, nil
@@ -441,14 +440,14 @@ func (f *folder) pull() (success bool, err error) {
// Pulling failed, try again later.
delay := f.pullPause + time.Since(startTime)
f.sl.Info("Folder failed to sync, will be retried", slog.String("wait", stringutil.NiceDurationString(delay)))
f.sl.InfoContext(ctx, "Folder failed to sync, will be retried", slog.String("wait", stringutil.NiceDurationString(delay)))
f.pullFailTimer.Reset(delay)
return false, err
}
func (f *folder) scanSubdirs(subDirs []string) error {
l.Debugf("%v scanning", f)
func (f *folder) scanSubdirs(ctx context.Context, subDirs []string) error {
f.sl.DebugContext(ctx, "Scanning")
oldHash := f.ignores.Hash()
@@ -456,14 +455,14 @@ func (f *folder) scanSubdirs(subDirs []string) error {
if err != nil {
return err
}
f.setError(nil)
f.setError(ctx, nil)
// Check on the way out if the ignore patterns changed as part of scanning
// this folder. If they did we should schedule a pull of the folder so that
// we request things we might have suddenly become unignored and so on.
defer func() {
if f.ignores.Hash() != oldHash {
l.Debugln("Folder", f.Description(), "ignore patterns change detected while scanning; triggering puller")
f.sl.DebugContext(ctx, "Ignore patterns change detected while scanning; triggering puller")
f.ignoresUpdated()
f.SchedulePull()
}
@@ -472,15 +471,15 @@ func (f *folder) scanSubdirs(subDirs []string) error {
f.setState(FolderScanWaiting)
defer f.setState(FolderIdle)
if err := f.ioLimiter.TakeWithContext(f.ctx, 1); err != nil {
if err := f.ioLimiter.TakeWithContext(ctx, 1); err != nil {
return err
}
defer f.ioLimiter.Give(1)
metricFolderScans.WithLabelValues(f.ID).Inc()
ctx, cancel := context.WithCancel(f.ctx)
scanCtx, cancel := context.WithCancel(ctx)
defer cancel()
go addTimeUntilCancelled(ctx, metricFolderScanSeconds.WithLabelValues(f.ID))
go addTimeUntilCancelled(scanCtx, metricFolderScanSeconds.WithLabelValues(f.ID))
for i := range subDirs {
sub := osutil.NativeFilename(subDirs[i])
@@ -512,13 +511,13 @@ func (f *folder) scanSubdirs(subDirs []string) error {
// changes.
changes := 0
defer func() {
l.Debugf("%v finished scanning, detected %v changes", f, changes)
f.sl.DebugContext(ctx, "Finished scanning", slog.Int("changes", changes))
if changes > 0 {
f.SchedulePull()
}
}()
changesHere, err := f.scanSubdirsChangedAndNew(subDirs, batch)
changesHere, err := f.scanSubdirsChangedAndNew(ctx, subDirs, batch)
changes += changesHere
if err != nil {
return err
@@ -537,7 +536,7 @@ func (f *folder) scanSubdirs(subDirs []string) error {
// Do a scan of the database for each prefix, to check for deleted and
// ignored files.
changesHere, err = f.scanSubdirsDeletedAndIgnored(subDirs, batch)
changesHere, err = f.scanSubdirsDeletedAndIgnored(ctx, subDirs, batch)
changes += changesHere
if err != nil {
return err
@@ -566,7 +565,7 @@ func (f *folder) newScanBatch() *scanBatch {
}
b.updateBatch = NewFileInfoBatch(func(fs []protocol.FileInfo) error {
if err := b.f.getHealthErrorWithoutIgnores(); err != nil {
l.Debugf("Stopping scan of folder %s due to: %s", b.f.Description(), err)
b.f.sl.Debug("Stopping scan due to folder error", slogutil.Error(err))
return err
}
b.f.updateLocalsFromScanning(fs)
@@ -627,7 +626,7 @@ func (b *scanBatch) Update(fi protocol.FileInfo) (bool, error) {
// Our item is deleted and the global item is our own receive only
// file. No point in keeping track of that.
b.Remove(fi.Name)
l.Debugf("%v scanning: deleting deleted receive-only local-changed file: %v", b.f, fi)
b.f.sl.Debug("Deleting deleted receive-only local-changed file", slogutil.FilePath(fi.Name))
return true, nil
}
case (b.f.Type == config.FolderTypeReceiveOnly || b.f.Type == config.FolderTypeReceiveEncrypted) &&
@@ -640,19 +639,19 @@ func (b *scanBatch) Update(fi protocol.FileInfo) (bool, error) {
IgnoreXattrs: !b.f.SyncXattrs && !b.f.SendXattrs,
}):
// What we have locally is equivalent to the global file.
l.Debugf("%v scanning: Merging identical locally changed item with global: %v", b.f, fi)
b.f.sl.Debug("Merging identical locally changed item with global", slogutil.FilePath(fi.Name))
fi = gf
}
b.updateBatch.Append(fi)
return true, nil
}
func (f *folder) scanSubdirsChangedAndNew(subDirs []string, batch *scanBatch) (int, error) {
func (f *folder) scanSubdirsChangedAndNew(ctx context.Context, subDirs []string, batch *scanBatch) (int, error) {
changes := 0
// If we return early e.g. due to a folder health error, the scan needs
// to be cancelled.
scanCtx, scanCancel := context.WithCancel(f.ctx)
scanCtx, scanCancel := context.WithCancel(ctx)
defer scanCancel()
scanConfig := scanner.Config{
@@ -706,7 +705,7 @@ func (f *folder) scanSubdirsChangedAndNew(subDirs []string, batch *scanBatch) (i
switch f.Type {
case config.FolderTypeReceiveOnly, config.FolderTypeReceiveEncrypted:
default:
if nf, ok := f.findRename(res.File, alreadyUsedOrExisting); ok {
if nf, ok := f.findRename(ctx, res.File, alreadyUsedOrExisting); ok {
if ok, err := batch.Update(nf); err != nil {
return 0, err
} else if ok {
@@ -719,7 +718,7 @@ func (f *folder) scanSubdirsChangedAndNew(subDirs []string, batch *scanBatch) (i
return changes, nil
}
func (f *folder) scanSubdirsDeletedAndIgnored(subDirs []string, batch *scanBatch) (int, error) {
func (f *folder) scanSubdirsDeletedAndIgnored(ctx context.Context, subDirs []string, batch *scanBatch) (int, error) {
var toIgnore []protocol.FileInfo
ignoredParent := ""
changes := 0
@@ -732,7 +731,7 @@ outer:
}
select {
case <-f.ctx.Done():
case <-ctx.Done():
break outer
default:
}
@@ -743,7 +742,7 @@ outer:
if ignoredParent != "" && !fs.IsParent(fi.Name, ignoredParent) {
for _, file := range toIgnore {
l.Debugln("marking file as ignored", file)
f.sl.DebugContext(ctx, "Marking file as ignored", slogutil.FilePath(file.Name))
nf := file
nf.SetIgnored()
if ok, err := batch.Update(nf); err != nil {
@@ -775,7 +774,7 @@ outer:
continue
}
l.Debugln("marking file as ignored", fi)
f.sl.DebugContext(ctx, "Marking file as ignored", slogutil.FilePath(fi.Name))
nf := fi
nf.SetIgnored()
if ok, err := batch.Update(nf); err != nil {
@@ -810,7 +809,7 @@ outer:
// sure the file gets in sync on the following pull.
nf.Version = protocol.Vector{}
}
l.Debugln("marking file as deleted", nf)
f.sl.DebugContext(ctx, "Marking file as deleted", slogutil.FilePath(nf.Name))
if ok, err := batch.Update(nf); err != nil {
return 0, err
} else if ok {
@@ -824,13 +823,13 @@ outer:
return 0, err
case !ok:
case gf.IsReceiveOnlyChanged():
l.Debugln("removing deleted, receive-only item that is globally receive-only from db", fi)
f.sl.DebugContext(ctx, "Removing deleted receive-only item that is globally receive-only from db", slogutil.FilePath(fi.Name))
batch.Remove(fi.Name)
changes++
case gf.IsDeleted():
// Our item is deleted and the global item is deleted too. We just
// pretend it is a normal deleted file (nobody cares about that).
l.Debugf("%v scanning: Marking globally deleted item as not locally changed: %v", f, fi.Name)
f.sl.DebugContext(ctx, "Marking globally deleted item as not locally changed", slogutil.FilePath(fi.Name))
fi.LocalFlags &^= protocol.FlagLocalReceiveOnly
if ok, err := batch.Update(fi); err != nil {
return 0, err
@@ -842,7 +841,7 @@ outer:
// No need to bump the version for a file that was and is
// deleted and just the folder type/local flags changed.
fi.LocalFlags &^= protocol.FlagLocalReceiveOnly
l.Debugln("removing receive-only flag on deleted item", fi)
f.sl.DebugContext(ctx, "Removing receive-only flag on deleted item", slogutil.FilePath(fi.Name))
if ok, err := batch.Update(fi); err != nil {
return 0, err
} else if ok {
@@ -853,14 +852,14 @@ outer:
}
select {
case <-f.ctx.Done():
return changes, f.ctx.Err()
case <-ctx.Done():
return changes, ctx.Err()
default:
}
if len(toIgnore) > 0 {
for _, file := range toIgnore {
l.Debugln("marking file as ignored", file)
f.sl.DebugContext(ctx, "Marking file as ignored", slogutil.FilePath(file.Name))
nf := file
nf.SetIgnored()
if ok, err := batch.Update(nf); err != nil {
@@ -879,7 +878,7 @@ outer:
return changes, nil
}
func (f *folder) findRename(file protocol.FileInfo, alreadyUsedOrExisting map[string]struct{}) (protocol.FileInfo, bool) {
func (f *folder) findRename(ctx context.Context, file protocol.FileInfo, alreadyUsedOrExisting map[string]struct{}) (protocol.FileInfo, bool) {
if len(file.Blocks) == 0 || file.Size == 0 {
return protocol.FileInfo{}, false
}
@@ -894,7 +893,7 @@ loop:
}
select {
case <-f.ctx.Done():
case <-ctx.Done():
break loop
default:
}
@@ -943,16 +942,16 @@ loop:
return nf, found
}
func (f *folder) scanTimerFired() error {
err := f.scanSubdirs(nil)
func (f *folder) scanTimerFired(ctx context.Context) error {
err := f.scanSubdirs(ctx, nil)
select {
case <-f.initialScanFinished:
default:
if err != nil {
f.sl.Error("Failed initial scan", slogutil.Error(err))
f.sl.ErrorContext(ctx, "Failed initial scan", slogutil.Error(err))
} else {
f.sl.Info("Completed initial scan")
f.sl.InfoContext(ctx, "Completed initial scan")
}
close(f.initialScanFinished)
}
@@ -962,19 +961,19 @@ func (f *folder) scanTimerFired() error {
return err
}
func (f *folder) versionCleanupTimerFired() {
func (f *folder) versionCleanupTimerFired(ctx context.Context) {
f.setState(FolderCleanWaiting)
defer f.setState(FolderIdle)
if err := f.ioLimiter.TakeWithContext(f.ctx, 1); err != nil {
if err := f.ioLimiter.TakeWithContext(ctx, 1); err != nil {
return
}
defer f.ioLimiter.Give(1)
f.setState(FolderCleaning)
if err := f.versioner.Clean(f.ctx); err != nil {
f.sl.Warn("Failed to clean versions", slogutil.Error(err))
if err := f.versioner.Clean(ctx); err != nil {
f.sl.WarnContext(ctx, "Failed to clean versions", slogutil.Error(err))
}
f.versionCleanupTimer.Reset(f.versionCleanupInterval)
@@ -1008,21 +1007,21 @@ func (f *folder) scheduleWatchRestart() {
// restartWatch should only ever be called synchronously. If you want to use
// this asynchronously, you should probably use scheduleWatchRestart instead.
func (f *folder) restartWatch() error {
func (f *folder) restartWatch(ctx context.Context) error {
f.stopWatch()
f.startWatch()
return f.scanSubdirs(nil)
f.startWatch(ctx)
return f.scanSubdirs(ctx, nil)
}
// startWatch should only ever be called synchronously. If you want to use
// this asynchronously, you should probably use scheduleWatchRestart instead.
func (f *folder) startWatch() {
ctx, cancel := context.WithCancel(f.ctx)
func (f *folder) startWatch(ctx context.Context) {
watchCtx, cancel := context.WithCancel(ctx)
f.watchMut.Lock()
f.watchChan = make(chan []string)
f.watchCancel = cancel
f.watchMut.Unlock()
go f.monitorWatch(ctx)
go f.monitorWatch(watchCtx)
}
// monitorWatch starts the filesystem watching and retries every minute on failure.
@@ -1066,7 +1065,7 @@ func (f *folder) monitorWatch(ctx context.Context) {
}
lastWatch = time.Now()
watchaggregator.Aggregate(aggrCtx, eventChan, f.watchChan, f.FolderConfiguration, f.model.cfg, f.evLogger)
l.Debugln("Started filesystem watcher for folder", f.Description())
f.sl.DebugContext(ctx, "Started filesystem watcher")
case err = <-errChan:
var next time.Duration
if dur := time.Since(lastWatch); dur > pause {
@@ -1148,9 +1147,9 @@ func (f *folder) scanOnWatchErr() {
}
}
func (f *folder) setError(err error) {
func (f *folder) setError(ctx context.Context, err error) {
select {
case <-f.ctx.Done():
case <-ctx.Done():
return
default:
}
@@ -1162,12 +1161,12 @@ func (f *folder) setError(err error) {
if err != nil {
if oldErr == nil {
f.sl.Warn("Error on folder", slogutil.Error(err))
f.sl.WarnContext(ctx, "Error on folder", slogutil.Error(err))
} else {
f.sl.Info("Folder error changed", slogutil.Error(err), slog.Any("previously", oldErr))
f.sl.InfoContext(ctx, "Folder error changed", slogutil.Error(err), slog.Any("previously", oldErr))
}
} else {
f.sl.Info("Folder error cleared")
f.sl.InfoContext(ctx, "Folder error cleared")
f.SchedulePull()
}
@@ -1324,7 +1323,7 @@ func (f *folder) emitDiskChangeEvents(fs []protocol.FileInfo, typeOfEvent events
}
}
func (f *folder) handleForcedRescans() error {
func (f *folder) handleForcedRescans(ctx context.Context) error {
f.forcedRescanPathsMut.Lock()
paths := make([]string, 0, len(f.forcedRescanPaths))
for path := range f.forcedRescanPaths {
@@ -1360,7 +1359,7 @@ func (f *folder) handleForcedRescans() error {
return err
}
return f.scanSubdirs(paths)
return f.scanSubdirs(ctx, paths)
}
// The exists function is expected to return true for all known paths
@@ -1377,7 +1376,7 @@ func unifySubs(dirs []string, exists func(dir string) bool) []string {
for i := 0; i < len(dirs); {
dir, err := fs.Canonicalize(dirs[i])
if err != nil {
l.Debugf("Skipping %v for scan: %s", dirs[i], err)
slog.Debug("Skipping directory for scan", slog.String("dir", dirs[i]), slogutil.Error(err))
dirs = append(dirs[:i], dirs[i+1:]...)
continue
}