folder.go 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149
  1. // Copyright (C) 2014 The Syncthing Authors.
  2. //
  3. // This Source Code Form is subject to the terms of the Mozilla Public
  4. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  5. // You can obtain one at https://mozilla.org/MPL/2.0/.
  6. package model
  7. import (
  8. "context"
  9. "fmt"
  10. "math/rand"
  11. "path/filepath"
  12. "sort"
  13. "sync/atomic"
  14. "time"
  15. "github.com/pkg/errors"
  16. "github.com/syncthing/syncthing/lib/config"
  17. "github.com/syncthing/syncthing/lib/db"
  18. "github.com/syncthing/syncthing/lib/events"
  19. "github.com/syncthing/syncthing/lib/fs"
  20. "github.com/syncthing/syncthing/lib/ignore"
  21. "github.com/syncthing/syncthing/lib/locations"
  22. "github.com/syncthing/syncthing/lib/osutil"
  23. "github.com/syncthing/syncthing/lib/protocol"
  24. "github.com/syncthing/syncthing/lib/scanner"
  25. "github.com/syncthing/syncthing/lib/stats"
  26. "github.com/syncthing/syncthing/lib/sync"
  27. "github.com/syncthing/syncthing/lib/util"
  28. "github.com/syncthing/syncthing/lib/versioner"
  29. "github.com/syncthing/syncthing/lib/watchaggregator"
  30. "github.com/thejerf/suture"
  31. )
  32. type folder struct {
  33. suture.Service
  34. stateTracker
  35. config.FolderConfiguration
  36. *stats.FolderStatisticsReference
  37. ioLimiter *byteSemaphore
  38. localFlags uint32
  39. model *model
  40. shortID protocol.ShortID
  41. fset *db.FileSet
  42. ignores *ignore.Matcher
  43. ctx context.Context
  44. scanInterval time.Duration
  45. scanTimer *time.Timer
  46. scanDelay chan time.Duration
  47. initialScanFinished chan struct{}
  48. scanErrors []FileError
  49. scanErrorsMut sync.Mutex
  50. versionCleanupInterval time.Duration
  51. versionCleanupTimer *time.Timer
  52. pullScheduled chan struct{}
  53. pullPause time.Duration
  54. pullFailTimer *time.Timer
  55. doInSyncChan chan syncRequest
  56. forcedRescanRequested chan struct{}
  57. forcedRescanPaths map[string]struct{}
  58. forcedRescanPathsMut sync.Mutex
  59. watchCancel context.CancelFunc
  60. watchChan chan []string
  61. restartWatchChan chan struct{}
  62. watchErr error
  63. watchMut sync.Mutex
  64. puller puller
  65. versioner versioner.Versioner
  66. }
  67. type syncRequest struct {
  68. fn func() error
  69. err chan error
  70. }
  71. type puller interface {
  72. pull() bool // true when successful and should not be retried
  73. }
  74. func newFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, evLogger events.Logger, ioLimiter *byteSemaphore, ver versioner.Versioner) folder {
  75. f := folder{
  76. stateTracker: newStateTracker(cfg.ID, evLogger),
  77. FolderConfiguration: cfg,
  78. FolderStatisticsReference: stats.NewFolderStatisticsReference(model.db, cfg.ID),
  79. ioLimiter: ioLimiter,
  80. model: model,
  81. shortID: model.shortID,
  82. fset: fset,
  83. ignores: ignores,
  84. scanInterval: time.Duration(cfg.RescanIntervalS) * time.Second,
  85. scanTimer: time.NewTimer(0), // The first scan should be done immediately.
  86. scanDelay: make(chan time.Duration),
  87. initialScanFinished: make(chan struct{}),
  88. scanErrorsMut: sync.NewMutex(),
  89. versionCleanupInterval: time.Duration(cfg.Versioning.CleanupIntervalS) * time.Second,
  90. versionCleanupTimer: time.NewTimer(time.Duration(cfg.Versioning.CleanupIntervalS) * time.Second),
  91. pullScheduled: make(chan struct{}, 1), // This needs to be 1-buffered so that we queue a pull if we're busy when it comes.
  92. doInSyncChan: make(chan syncRequest),
  93. forcedRescanRequested: make(chan struct{}, 1),
  94. forcedRescanPaths: make(map[string]struct{}),
  95. forcedRescanPathsMut: sync.NewMutex(),
  96. watchCancel: func() {},
  97. restartWatchChan: make(chan struct{}, 1),
  98. watchMut: sync.NewMutex(),
  99. versioner: ver,
  100. }
  101. f.pullPause = f.pullBasePause()
  102. f.pullFailTimer = time.NewTimer(0)
  103. <-f.pullFailTimer.C
  104. return f
  105. }
  106. func (f *folder) serve(ctx context.Context) {
  107. atomic.AddInt32(&f.model.foldersRunning, 1)
  108. defer atomic.AddInt32(&f.model.foldersRunning, -1)
  109. f.ctx = ctx
  110. l.Debugln(f, "starting")
  111. defer l.Debugln(f, "exiting")
  112. defer func() {
  113. f.scanTimer.Stop()
  114. f.versionCleanupTimer.Stop()
  115. f.setState(FolderIdle)
  116. }()
  117. if f.FSWatcherEnabled && f.getHealthErrorAndLoadIgnores() == nil {
  118. f.startWatch()
  119. }
  120. // If we're configured to not do version cleanup, or we don't have a
  121. // versioner, cancel and drain that timer now.
  122. if f.versionCleanupInterval == 0 || f.versioner == nil {
  123. if !f.versionCleanupTimer.Stop() {
  124. <-f.versionCleanupTimer.C
  125. }
  126. }
  127. initialCompleted := f.initialScanFinished
  128. for {
  129. select {
  130. case <-f.ctx.Done():
  131. return
  132. case <-f.pullScheduled:
  133. f.pull()
  134. case <-f.pullFailTimer.C:
  135. if !f.pull() && f.pullPause < 60*f.pullBasePause() {
  136. // Back off from retrying to pull
  137. f.pullPause *= 2
  138. }
  139. case <-initialCompleted:
  140. // Initial scan has completed, we should do a pull
  141. initialCompleted = nil // never hit this case again
  142. f.pull()
  143. case <-f.forcedRescanRequested:
  144. f.handleForcedRescans()
  145. case <-f.scanTimer.C:
  146. l.Debugln(f, "Scanning due to timer")
  147. f.scanTimerFired()
  148. case req := <-f.doInSyncChan:
  149. l.Debugln(f, "Running something due to request")
  150. req.err <- req.fn()
  151. case next := <-f.scanDelay:
  152. l.Debugln(f, "Delaying scan")
  153. f.scanTimer.Reset(next)
  154. case fsEvents := <-f.watchChan:
  155. l.Debugln(f, "Scan due to watcher")
  156. f.scanSubdirs(fsEvents)
  157. case <-f.restartWatchChan:
  158. l.Debugln(f, "Restart watcher")
  159. f.restartWatch()
  160. case <-f.versionCleanupTimer.C:
  161. l.Debugln(f, "Doing version cleanup")
  162. f.versionCleanupTimerFired()
  163. }
  164. }
  165. }
  166. func (f *folder) BringToFront(string) {}
  167. func (f *folder) Override() {}
  168. func (f *folder) Revert() {}
  169. func (f *folder) DelayScan(next time.Duration) {
  170. f.Delay(next)
  171. }
  172. func (f *folder) ignoresUpdated() {
  173. if f.FSWatcherEnabled {
  174. f.scheduleWatchRestart()
  175. }
  176. }
  177. func (f *folder) SchedulePull() {
  178. select {
  179. case f.pullScheduled <- struct{}{}:
  180. default:
  181. // We might be busy doing a pull and thus not reading from this
  182. // channel. The channel is 1-buffered, so one notification will be
  183. // queued to ensure we recheck after the pull, but beyond that we must
  184. // make sure to not block index receiving.
  185. }
  186. }
  187. func (f *folder) Jobs(_, _ int) ([]string, []string, int) {
  188. return nil, nil, 0
  189. }
  190. func (f *folder) Scan(subdirs []string) error {
  191. <-f.initialScanFinished
  192. return f.doInSync(func() error { return f.scanSubdirs(subdirs) })
  193. }
  194. // doInSync allows to run functions synchronously in folder.serve from exported,
  195. // asynchronously called methods.
  196. func (f *folder) doInSync(fn func() error) error {
  197. req := syncRequest{
  198. fn: fn,
  199. err: make(chan error, 1),
  200. }
  201. select {
  202. case f.doInSyncChan <- req:
  203. return <-req.err
  204. case <-f.ctx.Done():
  205. return f.ctx.Err()
  206. }
  207. }
  208. func (f *folder) Reschedule() {
  209. if f.scanInterval == 0 {
  210. return
  211. }
  212. // Sleep a random time between 3/4 and 5/4 of the configured interval.
  213. sleepNanos := (f.scanInterval.Nanoseconds()*3 + rand.Int63n(2*f.scanInterval.Nanoseconds())) / 4
  214. interval := time.Duration(sleepNanos) * time.Nanosecond
  215. l.Debugln(f, "next rescan in", interval)
  216. f.scanTimer.Reset(interval)
  217. }
  218. func (f *folder) Delay(next time.Duration) {
  219. f.scanDelay <- next
  220. }
  221. func (f *folder) getHealthErrorAndLoadIgnores() error {
  222. if err := f.getHealthErrorWithoutIgnores(); err != nil {
  223. return err
  224. }
  225. if err := f.ignores.Load(".stignore"); err != nil && !fs.IsNotExist(err) {
  226. return errors.Wrap(err, "loading ignores")
  227. }
  228. return nil
  229. }
  230. func (f *folder) getHealthErrorWithoutIgnores() error {
  231. // Check for folder errors, with the most serious and specific first and
  232. // generic ones like out of space on the home disk later.
  233. if err := f.CheckPath(); err != nil {
  234. return err
  235. }
  236. dbPath := locations.Get(locations.Database)
  237. if usage, err := fs.NewFilesystem(fs.FilesystemTypeBasic, dbPath).Usage("."); err == nil {
  238. if err = config.CheckFreeSpace(f.model.cfg.Options().MinHomeDiskFree, usage); err != nil {
  239. return errors.Wrapf(err, "insufficient space on disk for database (%v)", dbPath)
  240. }
  241. }
  242. return nil
  243. }
  244. func (f *folder) pull() (success bool) {
  245. f.pullFailTimer.Stop()
  246. select {
  247. case <-f.pullFailTimer.C:
  248. default:
  249. }
  250. select {
  251. case <-f.initialScanFinished:
  252. default:
  253. // Once the initial scan finished, a pull will be scheduled
  254. return true
  255. }
  256. defer func() {
  257. if success {
  258. // We're good, reset the pause interval.
  259. f.pullPause = f.pullBasePause()
  260. }
  261. }()
  262. // If there is nothing to do, don't even enter sync-waiting state.
  263. abort := true
  264. snap := f.fset.Snapshot()
  265. snap.WithNeed(protocol.LocalDeviceID, func(intf protocol.FileIntf) bool {
  266. abort = false
  267. return false
  268. })
  269. snap.Release()
  270. if abort {
  271. return true
  272. }
  273. // Abort early (before acquiring a token) if there's a folder error
  274. err := f.getHealthErrorWithoutIgnores()
  275. f.setError(err)
  276. if err != nil {
  277. l.Debugln("Skipping pull of", f.Description(), "due to folder error:", err)
  278. return false
  279. }
  280. f.setState(FolderSyncWaiting)
  281. if err := f.ioLimiter.takeWithContext(f.ctx, 1); err != nil {
  282. f.setError(err)
  283. return true
  284. }
  285. defer f.ioLimiter.give(1)
  286. startTime := time.Now()
  287. // Check if the ignore patterns changed.
  288. oldHash := f.ignores.Hash()
  289. defer func() {
  290. if f.ignores.Hash() != oldHash {
  291. f.ignoresUpdated()
  292. }
  293. }()
  294. err = f.getHealthErrorAndLoadIgnores()
  295. f.setError(err)
  296. if err != nil {
  297. l.Debugln("Skipping pull of", f.Description(), "due to folder error:", err)
  298. return false
  299. }
  300. success = f.puller.pull()
  301. if success {
  302. return true
  303. }
  304. // Pulling failed, try again later.
  305. delay := f.pullPause + time.Since(startTime)
  306. l.Infof("Folder %v isn't making sync progress - retrying in %v.", f.Description(), util.NiceDurationString(delay))
  307. f.pullFailTimer.Reset(delay)
  308. return false
  309. }
  310. func (f *folder) scanSubdirs(subDirs []string) error {
  311. oldHash := f.ignores.Hash()
  312. err := f.getHealthErrorAndLoadIgnores()
  313. f.setError(err)
  314. if err != nil {
  315. // If there is a health error we set it as the folder error. We do not
  316. // clear the folder error if there is no health error, as there might be
  317. // an *other* folder error (failed to load ignores, for example). Hence
  318. // we do not use the CheckHealth() convenience function here.
  319. return err
  320. }
  321. // Check on the way out if the ignore patterns changed as part of scanning
  322. // this folder. If they did we should schedule a pull of the folder so that
  323. // we request things we might have suddenly become unignored and so on.
  324. defer func() {
  325. if f.ignores.Hash() != oldHash {
  326. l.Debugln("Folder", f.Description(), "ignore patterns change detected while scanning; triggering puller")
  327. f.ignoresUpdated()
  328. f.SchedulePull()
  329. }
  330. }()
  331. f.setState(FolderScanWaiting)
  332. defer f.setState(FolderIdle)
  333. if err := f.ioLimiter.takeWithContext(f.ctx, 1); err != nil {
  334. return err
  335. }
  336. defer f.ioLimiter.give(1)
  337. for i := range subDirs {
  338. sub := osutil.NativeFilename(subDirs[i])
  339. if sub == "" {
  340. // A blank subdirs means to scan the entire folder. We can trim
  341. // the subDirs list and go on our way.
  342. subDirs = nil
  343. break
  344. }
  345. subDirs[i] = sub
  346. }
  347. snap := f.fset.Snapshot()
  348. // We release explicitly later in this function, however we might exit early
  349. // and it's ok to release twice.
  350. defer snap.Release()
  351. // Clean the list of subitems to ensure that we start at a known
  352. // directory, and don't scan subdirectories of things we've already
  353. // scanned.
  354. subDirs = unifySubs(subDirs, func(file string) bool {
  355. _, ok := snap.Get(protocol.LocalDeviceID, file)
  356. return ok
  357. })
  358. f.setState(FolderScanning)
  359. // If we return early e.g. due to a folder health error, the scan needs
  360. // to be cancelled.
  361. scanCtx, scanCancel := context.WithCancel(f.ctx)
  362. defer scanCancel()
  363. mtimefs := f.fset.MtimeFS()
  364. fchan := scanner.Walk(scanCtx, scanner.Config{
  365. Folder: f.ID,
  366. Subs: subDirs,
  367. Matcher: f.ignores,
  368. TempLifetime: time.Duration(f.model.cfg.Options().KeepTemporariesH) * time.Hour,
  369. CurrentFiler: cFiler{snap},
  370. Filesystem: mtimefs,
  371. IgnorePerms: f.IgnorePerms,
  372. AutoNormalize: f.AutoNormalize,
  373. Hashers: f.model.numHashers(f.ID),
  374. ShortID: f.shortID,
  375. ProgressTickIntervalS: f.ScanProgressIntervalS,
  376. LocalFlags: f.localFlags,
  377. ModTimeWindow: f.ModTimeWindow(),
  378. EventLogger: f.evLogger,
  379. })
  380. batch := newFileInfoBatch(func(fs []protocol.FileInfo) error {
  381. if err := f.getHealthErrorWithoutIgnores(); err != nil {
  382. l.Debugf("Stopping scan of folder %s due to: %s", f.Description(), err)
  383. return err
  384. }
  385. f.updateLocalsFromScanning(fs)
  386. return nil
  387. })
  388. var batchAppend func(protocol.FileInfo, *db.Snapshot)
  389. // Resolve items which are identical with the global state.
  390. if f.localFlags&protocol.FlagLocalReceiveOnly == 0 {
  391. batchAppend = func(fi protocol.FileInfo, _ *db.Snapshot) {
  392. batch.append(fi)
  393. }
  394. } else {
  395. batchAppend = func(fi protocol.FileInfo, snap *db.Snapshot) {
  396. switch gf, ok := snap.GetGlobal(fi.Name); {
  397. case !ok:
  398. case gf.IsEquivalentOptional(fi, f.ModTimeWindow(), false, false, protocol.FlagLocalReceiveOnly):
  399. // What we have locally is equivalent to the global file.
  400. fi.Version = fi.Version.Merge(gf.Version)
  401. fallthrough
  402. case fi.IsDeleted() && gf.IsReceiveOnlyChanged():
  403. // Our item is deleted and the global item is our own
  404. // receive only file. We can't delete file infos, so
  405. // we just pretend it is a normal deleted file (nobody
  406. // cares about that).
  407. fi.LocalFlags &^= protocol.FlagLocalReceiveOnly
  408. }
  409. batch.append(fi)
  410. }
  411. }
  412. // Schedule a pull after scanning, but only if we actually detected any
  413. // changes.
  414. changes := 0
  415. defer func() {
  416. if changes > 0 {
  417. f.SchedulePull()
  418. }
  419. }()
  420. f.clearScanErrors(subDirs)
  421. alreadyUsed := make(map[string]struct{})
  422. for res := range fchan {
  423. if res.Err != nil {
  424. f.newScanError(res.Path, res.Err)
  425. continue
  426. }
  427. if err := batch.flushIfFull(); err != nil {
  428. return err
  429. }
  430. batchAppend(res.File, snap)
  431. changes++
  432. if f.localFlags&protocol.FlagLocalReceiveOnly == 0 {
  433. if nf, ok := f.findRename(snap, mtimefs, res.File, alreadyUsed); ok {
  434. batchAppend(nf, snap)
  435. changes++
  436. }
  437. }
  438. }
  439. if err := batch.flush(); err != nil {
  440. return err
  441. }
  442. if len(subDirs) == 0 {
  443. // If we have no specific subdirectories to traverse, set it to one
  444. // empty prefix so we traverse the entire folder contents once.
  445. subDirs = []string{""}
  446. }
  447. // Do a scan of the database for each prefix, to check for deleted and
  448. // ignored files.
  449. var toIgnore []db.FileInfoTruncated
  450. ignoredParent := ""
  451. snap.Release()
  452. snap = f.fset.Snapshot()
  453. defer snap.Release()
  454. for _, sub := range subDirs {
  455. var iterError error
  456. snap.WithPrefixedHaveTruncated(protocol.LocalDeviceID, sub, func(fi protocol.FileIntf) bool {
  457. select {
  458. case <-f.ctx.Done():
  459. return false
  460. default:
  461. }
  462. file := fi.(db.FileInfoTruncated)
  463. if err := batch.flushIfFull(); err != nil {
  464. iterError = err
  465. return false
  466. }
  467. if ignoredParent != "" && !fs.IsParent(file.Name, ignoredParent) {
  468. for _, file := range toIgnore {
  469. l.Debugln("marking file as ignored", file)
  470. nf := file.ConvertToIgnoredFileInfo(f.shortID)
  471. batchAppend(nf, snap)
  472. changes++
  473. if err := batch.flushIfFull(); err != nil {
  474. iterError = err
  475. return false
  476. }
  477. }
  478. toIgnore = toIgnore[:0]
  479. ignoredParent = ""
  480. }
  481. switch ignored := f.ignores.Match(file.Name).IsIgnored(); {
  482. case file.IsIgnored() && ignored:
  483. return true
  484. case !file.IsIgnored() && ignored:
  485. // File was not ignored at last pass but has been ignored.
  486. if file.IsDirectory() {
  487. // Delay ignoring as a child might be unignored.
  488. toIgnore = append(toIgnore, file)
  489. if ignoredParent == "" {
  490. // If the parent wasn't ignored already, set
  491. // this path as the "highest" ignored parent
  492. ignoredParent = file.Name
  493. }
  494. return true
  495. }
  496. l.Debugln("marking file as ignored", file)
  497. nf := file.ConvertToIgnoredFileInfo(f.shortID)
  498. batchAppend(nf, snap)
  499. changes++
  500. case file.IsIgnored() && !ignored:
  501. // Successfully scanned items are already un-ignored during
  502. // the scan, so check whether it is deleted.
  503. fallthrough
  504. case !file.IsIgnored() && !file.IsDeleted() && !file.IsUnsupported():
  505. // The file is not ignored, deleted or unsupported. Lets check if
  506. // it's still here. Simply stat:ing it wont do as there are
  507. // tons of corner cases (e.g. parent dir->symlink, missing
  508. // permissions)
  509. if !osutil.IsDeleted(mtimefs, file.Name) {
  510. if ignoredParent != "" {
  511. // Don't ignore parents of this not ignored item
  512. toIgnore = toIgnore[:0]
  513. ignoredParent = ""
  514. }
  515. return true
  516. }
  517. nf := file.ConvertToDeletedFileInfo(f.shortID)
  518. nf.LocalFlags = f.localFlags
  519. if file.ShouldConflict() {
  520. // We do not want to override the global version with
  521. // the deleted file. Setting to an empty version makes
  522. // sure the file gets in sync on the following pull.
  523. nf.Version = protocol.Vector{}
  524. }
  525. // Check for deleted, locally changed items that noone else has.
  526. if f.localFlags&protocol.FlagLocalReceiveOnly != 0 && len(snap.Availability(file.Name)) == 0 {
  527. file.LocalFlags &^= protocol.FlagLocalReceiveOnly
  528. }
  529. l.Debugln("marking file as deleted", nf)
  530. batchAppend(nf, snap)
  531. changes++
  532. case file.IsDeleted() && file.IsReceiveOnlyChanged() && f.localFlags&protocol.FlagLocalReceiveOnly != 0 && len(snap.Availability(file.Name)) == 0:
  533. file.Version = protocol.Vector{}
  534. file.LocalFlags &^= protocol.FlagLocalReceiveOnly
  535. l.Debugln("marking deleted item that doesn't exist anywhere as not receive-only", file)
  536. batchAppend(file.ConvertDeletedToFileInfo(), snap)
  537. changes++
  538. case file.IsDeleted() && file.IsReceiveOnlyChanged() && f.Type != config.FolderTypeReceiveOnly:
  539. // No need to bump the version for a file that was and is
  540. // deleted and just the folder type/local flags changed.
  541. file.LocalFlags &^= protocol.FlagLocalReceiveOnly
  542. l.Debugln("removing receive-only flag on deleted item", file)
  543. batchAppend(file.ConvertDeletedToFileInfo(), snap)
  544. changes++
  545. }
  546. return true
  547. })
  548. select {
  549. case <-f.ctx.Done():
  550. return f.ctx.Err()
  551. default:
  552. }
  553. if iterError == nil && len(toIgnore) > 0 {
  554. for _, file := range toIgnore {
  555. l.Debugln("marking file as ignored", f)
  556. nf := file.ConvertToIgnoredFileInfo(f.shortID)
  557. batchAppend(nf, snap)
  558. changes++
  559. if iterError = batch.flushIfFull(); iterError != nil {
  560. break
  561. }
  562. }
  563. toIgnore = toIgnore[:0]
  564. }
  565. if iterError != nil {
  566. return iterError
  567. }
  568. }
  569. if err := batch.flush(); err != nil {
  570. return err
  571. }
  572. f.ScanCompleted()
  573. return nil
  574. }
  575. func (f *folder) findRename(snap *db.Snapshot, mtimefs fs.Filesystem, file protocol.FileInfo, alreadyUsed map[string]struct{}) (protocol.FileInfo, bool) {
  576. if len(file.Blocks) == 0 || file.Size == 0 {
  577. return protocol.FileInfo{}, false
  578. }
  579. found := false
  580. nf := protocol.FileInfo{}
  581. snap.WithBlocksHash(file.BlocksHash, func(ifi protocol.FileIntf) bool {
  582. fi := ifi.(protocol.FileInfo)
  583. select {
  584. case <-f.ctx.Done():
  585. return false
  586. default:
  587. }
  588. if _, ok := alreadyUsed[fi.Name]; ok {
  589. return true
  590. }
  591. if fi.ShouldConflict() {
  592. return true
  593. }
  594. if f.ignores.Match(fi.Name).IsIgnored() {
  595. return true
  596. }
  597. // Only check the size.
  598. // No point checking block equality, as that uses BlocksHash comparison if that is set (which it will be).
  599. // No point checking BlocksHash comparison as WithBlocksHash already does that.
  600. if file.Size != fi.Size {
  601. return true
  602. }
  603. if !osutil.IsDeleted(mtimefs, fi.Name) {
  604. return true
  605. }
  606. alreadyUsed[fi.Name] = struct{}{}
  607. nf = fi
  608. nf.SetDeleted(f.shortID)
  609. nf.LocalFlags = f.localFlags
  610. found = true
  611. return false
  612. })
  613. return nf, found
  614. }
  615. func (f *folder) scanTimerFired() {
  616. err := f.scanSubdirs(nil)
  617. select {
  618. case <-f.initialScanFinished:
  619. default:
  620. status := "Completed"
  621. if err != nil {
  622. status = "Failed"
  623. }
  624. l.Infoln(status, "initial scan of", f.Type.String(), "folder", f.Description())
  625. close(f.initialScanFinished)
  626. }
  627. f.Reschedule()
  628. }
  629. func (f *folder) versionCleanupTimerFired() {
  630. f.setState(FolderCleanWaiting)
  631. defer f.setState(FolderIdle)
  632. if err := f.ioLimiter.takeWithContext(f.ctx, 1); err != nil {
  633. return
  634. }
  635. defer f.ioLimiter.give(1)
  636. f.setState(FolderCleaning)
  637. if err := f.versioner.Clean(f.ctx); err != nil {
  638. l.Infoln("Failed to clean versions in %s: %v", f.Description(), err)
  639. }
  640. f.versionCleanupTimer.Reset(f.versionCleanupInterval)
  641. }
  642. func (f *folder) WatchError() error {
  643. f.watchMut.Lock()
  644. defer f.watchMut.Unlock()
  645. return f.watchErr
  646. }
  647. // stopWatch immediately aborts watching and may be called asynchronously
  648. func (f *folder) stopWatch() {
  649. f.watchMut.Lock()
  650. f.watchCancel()
  651. f.watchMut.Unlock()
  652. f.setWatchError(nil, 0)
  653. }
  654. // scheduleWatchRestart makes sure watching is restarted from the main for loop
  655. // in a folder's Serve and thus may be called asynchronously (e.g. when ignores change).
  656. func (f *folder) scheduleWatchRestart() {
  657. select {
  658. case f.restartWatchChan <- struct{}{}:
  659. default:
  660. // We might be busy doing a pull and thus not reading from this
  661. // channel. The channel is 1-buffered, so one notification will be
  662. // queued to ensure we recheck after the pull.
  663. }
  664. }
  665. // restartWatch should only ever be called synchronously. If you want to use
  666. // this asynchronously, you should probably use scheduleWatchRestart instead.
  667. func (f *folder) restartWatch() {
  668. f.stopWatch()
  669. f.startWatch()
  670. f.scanSubdirs(nil)
  671. }
  672. // startWatch should only ever be called synchronously. If you want to use
  673. // this asynchronously, you should probably use scheduleWatchRestart instead.
  674. func (f *folder) startWatch() {
  675. ctx, cancel := context.WithCancel(f.ctx)
  676. f.watchMut.Lock()
  677. f.watchChan = make(chan []string)
  678. f.watchCancel = cancel
  679. f.watchMut.Unlock()
  680. go f.monitorWatch(ctx)
  681. }
  682. // monitorWatch starts the filesystem watching and retries every minute on failure.
  683. // It should not be used except in startWatch.
  684. func (f *folder) monitorWatch(ctx context.Context) {
  685. failTimer := time.NewTimer(0)
  686. aggrCtx, aggrCancel := context.WithCancel(ctx)
  687. var err error
  688. var eventChan <-chan fs.Event
  689. var errChan <-chan error
  690. warnedOutside := false
  691. var lastWatch time.Time
  692. pause := time.Minute
  693. for {
  694. select {
  695. case <-failTimer.C:
  696. eventChan, errChan, err = f.Filesystem().Watch(".", f.ignores, ctx, f.IgnorePerms)
  697. // We do this once per minute initially increased to
  698. // max one hour in case of repeat failures.
  699. f.scanOnWatchErr()
  700. f.setWatchError(err, pause)
  701. if err != nil {
  702. failTimer.Reset(pause)
  703. if pause < 60*time.Minute {
  704. pause *= 2
  705. }
  706. continue
  707. }
  708. lastWatch = time.Now()
  709. watchaggregator.Aggregate(aggrCtx, eventChan, f.watchChan, f.FolderConfiguration, f.model.cfg, f.evLogger)
  710. l.Debugln("Started filesystem watcher for folder", f.Description())
  711. case err = <-errChan:
  712. var next time.Duration
  713. if dur := time.Since(lastWatch); dur > pause {
  714. pause = time.Minute
  715. next = 0
  716. } else {
  717. next = pause - dur
  718. if pause < 60*time.Minute {
  719. pause *= 2
  720. }
  721. }
  722. failTimer.Reset(next)
  723. f.setWatchError(err, next)
  724. // This error was previously a panic and should never occur, so generate
  725. // a warning, but don't do it repetitively.
  726. if !warnedOutside {
  727. if _, ok := err.(*fs.ErrWatchEventOutsideRoot); ok {
  728. l.Warnln(err)
  729. warnedOutside = true
  730. }
  731. }
  732. aggrCancel()
  733. errChan = nil
  734. aggrCtx, aggrCancel = context.WithCancel(ctx)
  735. case <-ctx.Done():
  736. return
  737. }
  738. }
  739. }
  740. // setWatchError sets the current error state of the watch and should be called
  741. // regardless of whether err is nil or not.
  742. func (f *folder) setWatchError(err error, nextTryIn time.Duration) {
  743. f.watchMut.Lock()
  744. prevErr := f.watchErr
  745. f.watchErr = err
  746. f.watchMut.Unlock()
  747. if err != prevErr {
  748. data := map[string]interface{}{
  749. "folder": f.ID,
  750. }
  751. if prevErr != nil {
  752. data["from"] = prevErr.Error()
  753. }
  754. if err != nil {
  755. data["to"] = err.Error()
  756. }
  757. f.evLogger.Log(events.FolderWatchStateChanged, data)
  758. }
  759. if err == nil {
  760. return
  761. }
  762. msg := fmt.Sprintf("Error while trying to start filesystem watcher for folder %s, trying again in %v: %v", f.Description(), nextTryIn, err)
  763. if prevErr != err {
  764. l.Infof(msg)
  765. return
  766. }
  767. l.Debugf(msg)
  768. }
  769. // scanOnWatchErr schedules a full scan immediately if an error occurred while watching.
  770. func (f *folder) scanOnWatchErr() {
  771. f.watchMut.Lock()
  772. err := f.watchErr
  773. f.watchMut.Unlock()
  774. if err != nil {
  775. f.Delay(0)
  776. }
  777. }
  778. func (f *folder) setError(err error) {
  779. select {
  780. case <-f.ctx.Done():
  781. return
  782. default:
  783. }
  784. _, _, oldErr := f.getState()
  785. if (err != nil && oldErr != nil && oldErr.Error() == err.Error()) || (err == nil && oldErr == nil) {
  786. return
  787. }
  788. if err != nil {
  789. if oldErr == nil {
  790. l.Warnf("Error on folder %s: %v", f.Description(), err)
  791. } else {
  792. l.Infof("Error on folder %s changed: %q -> %q", f.Description(), oldErr, err)
  793. }
  794. } else {
  795. l.Infoln("Cleared error on folder", f.Description())
  796. }
  797. if f.FSWatcherEnabled {
  798. if err != nil {
  799. f.stopWatch()
  800. } else {
  801. f.scheduleWatchRestart()
  802. }
  803. }
  804. f.stateTracker.setError(err)
  805. }
  806. func (f *folder) pullBasePause() time.Duration {
  807. if f.PullerPauseS == 0 {
  808. return defaultPullerPause
  809. }
  810. return time.Duration(f.PullerPauseS) * time.Second
  811. }
  812. func (f *folder) String() string {
  813. return fmt.Sprintf("%s/%s@%p", f.Type, f.folderID, f)
  814. }
  815. func (f *folder) newScanError(path string, err error) {
  816. f.scanErrorsMut.Lock()
  817. l.Infof("Scanner (folder %s, item %q): %v", f.Description(), path, err)
  818. f.scanErrors = append(f.scanErrors, FileError{
  819. Err: err.Error(),
  820. Path: path,
  821. })
  822. f.scanErrorsMut.Unlock()
  823. }
  824. func (f *folder) clearScanErrors(subDirs []string) {
  825. f.scanErrorsMut.Lock()
  826. defer f.scanErrorsMut.Unlock()
  827. if len(subDirs) == 0 {
  828. f.scanErrors = nil
  829. return
  830. }
  831. filtered := f.scanErrors[:0]
  832. outer:
  833. for _, fe := range f.scanErrors {
  834. for _, sub := range subDirs {
  835. if fe.Path == sub || fs.IsParent(fe.Path, sub) {
  836. continue outer
  837. }
  838. }
  839. filtered = append(filtered, fe)
  840. }
  841. f.scanErrors = filtered
  842. }
  843. func (f *folder) Errors() []FileError {
  844. f.scanErrorsMut.Lock()
  845. defer f.scanErrorsMut.Unlock()
  846. return append([]FileError{}, f.scanErrors...)
  847. }
  848. // ScheduleForceRescan marks the file such that it gets rehashed on next scan, and schedules a scan.
  849. func (f *folder) ScheduleForceRescan(path string) {
  850. f.forcedRescanPathsMut.Lock()
  851. f.forcedRescanPaths[path] = struct{}{}
  852. f.forcedRescanPathsMut.Unlock()
  853. select {
  854. case f.forcedRescanRequested <- struct{}{}:
  855. default:
  856. }
  857. }
  858. func (f *folder) updateLocalsFromScanning(fs []protocol.FileInfo) {
  859. f.updateLocals(fs)
  860. f.emitDiskChangeEvents(fs, events.LocalChangeDetected)
  861. }
  862. func (f *folder) updateLocalsFromPulling(fs []protocol.FileInfo) {
  863. f.updateLocals(fs)
  864. f.emitDiskChangeEvents(fs, events.RemoteChangeDetected)
  865. }
  866. func (f *folder) updateLocals(fs []protocol.FileInfo) {
  867. f.fset.Update(protocol.LocalDeviceID, fs)
  868. filenames := make([]string, len(fs))
  869. f.forcedRescanPathsMut.Lock()
  870. for i, file := range fs {
  871. filenames[i] = file.Name
  872. // No need to rescan a file that was changed since anyway.
  873. delete(f.forcedRescanPaths, file.Name)
  874. }
  875. f.forcedRescanPathsMut.Unlock()
  876. f.evLogger.Log(events.LocalIndexUpdated, map[string]interface{}{
  877. "folder": f.ID,
  878. "items": len(fs),
  879. "filenames": filenames,
  880. "version": f.fset.Sequence(protocol.LocalDeviceID),
  881. })
  882. }
  883. func (f *folder) emitDiskChangeEvents(fs []protocol.FileInfo, typeOfEvent events.EventType) {
  884. for _, file := range fs {
  885. if file.IsInvalid() {
  886. continue
  887. }
  888. objType := "file"
  889. action := "modified"
  890. if file.IsDeleted() {
  891. action = "deleted"
  892. }
  893. if file.IsSymlink() {
  894. objType = "symlink"
  895. } else if file.IsDirectory() {
  896. objType = "dir"
  897. }
  898. // Two different events can be fired here based on what EventType is passed into function
  899. f.evLogger.Log(typeOfEvent, map[string]string{
  900. "folder": f.ID,
  901. "folderID": f.ID, // incorrect, deprecated, kept for historical compliance
  902. "label": f.Label,
  903. "action": action,
  904. "type": objType,
  905. "path": filepath.FromSlash(file.Name),
  906. "modifiedBy": file.ModifiedBy.String(),
  907. })
  908. }
  909. }
  910. func (f *folder) handleForcedRescans() {
  911. f.forcedRescanPathsMut.Lock()
  912. paths := make([]string, 0, len(f.forcedRescanPaths))
  913. for path := range f.forcedRescanPaths {
  914. paths = append(paths, path)
  915. }
  916. f.forcedRescanPaths = make(map[string]struct{})
  917. f.forcedRescanPathsMut.Unlock()
  918. if len(paths) == 0 {
  919. return
  920. }
  921. batch := newFileInfoBatch(func(fs []protocol.FileInfo) error {
  922. f.fset.Update(protocol.LocalDeviceID, fs)
  923. return nil
  924. })
  925. snap := f.fset.Snapshot()
  926. for _, path := range paths {
  927. _ = batch.flushIfFull()
  928. fi, ok := snap.Get(protocol.LocalDeviceID, path)
  929. if !ok {
  930. continue
  931. }
  932. fi.SetMustRescan(f.shortID)
  933. batch.append(fi)
  934. }
  935. snap.Release()
  936. _ = batch.flush()
  937. _ = f.scanSubdirs(paths)
  938. }
  939. // The exists function is expected to return true for all known paths
  940. // (excluding "" and ".")
  941. func unifySubs(dirs []string, exists func(dir string) bool) []string {
  942. if len(dirs) == 0 {
  943. return nil
  944. }
  945. sort.Strings(dirs)
  946. if dirs[0] == "" || dirs[0] == "." || dirs[0] == string(fs.PathSeparator) {
  947. return nil
  948. }
  949. prev := "./" // Anything that can't be parent of a clean path
  950. for i := 0; i < len(dirs); {
  951. dir, err := fs.Canonicalize(dirs[i])
  952. if err != nil {
  953. l.Debugf("Skipping %v for scan: %s", dirs[i], err)
  954. dirs = append(dirs[:i], dirs[i+1:]...)
  955. continue
  956. }
  957. if dir == prev || fs.IsParent(dir, prev) {
  958. dirs = append(dirs[:i], dirs[i+1:]...)
  959. continue
  960. }
  961. parent := filepath.Dir(dir)
  962. for parent != "." && parent != string(fs.PathSeparator) && !exists(parent) {
  963. dir = parent
  964. parent = filepath.Dir(dir)
  965. }
  966. dirs[i] = dir
  967. prev = dir
  968. i++
  969. }
  970. return dirs
  971. }
  972. type cFiler struct {
  973. *db.Snapshot
  974. }
  975. // Implements scanner.CurrentFiler
  976. func (cf cFiler) CurrentFile(file string) (protocol.FileInfo, bool) {
  977. return cf.Get(protocol.LocalDeviceID, file)
  978. }