folder.go 27 KB

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