folder.go 26 KB

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