folder.go 32 KB

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