folder.go 25 KB

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