folder.go 34 KB

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