1
0

folder.go 26 KB

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