model.go 48 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792
  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 http://mozilla.org/MPL/2.0/.
  6. package model
  7. import (
  8. "bufio"
  9. "crypto/tls"
  10. "encoding/json"
  11. "errors"
  12. "fmt"
  13. "io"
  14. "io/ioutil"
  15. "net"
  16. "os"
  17. "path/filepath"
  18. "reflect"
  19. "runtime"
  20. "strings"
  21. stdsync "sync"
  22. "time"
  23. "github.com/syncthing/protocol"
  24. "github.com/syncthing/syncthing/internal/config"
  25. "github.com/syncthing/syncthing/internal/db"
  26. "github.com/syncthing/syncthing/internal/events"
  27. "github.com/syncthing/syncthing/internal/ignore"
  28. "github.com/syncthing/syncthing/internal/osutil"
  29. "github.com/syncthing/syncthing/internal/scanner"
  30. "github.com/syncthing/syncthing/internal/stats"
  31. "github.com/syncthing/syncthing/internal/symlinks"
  32. "github.com/syncthing/syncthing/internal/sync"
  33. "github.com/syncthing/syncthing/internal/versioner"
  34. "github.com/syndtr/goleveldb/leveldb"
  35. "github.com/thejerf/suture"
  36. )
  37. // How many files to send in each Index/IndexUpdate message.
  38. const (
  39. indexTargetSize = 250 * 1024 // Aim for making index messages no larger than 250 KiB (uncompressed)
  40. indexPerFileSize = 250 // Each FileInfo is approximately this big, in bytes, excluding BlockInfos
  41. indexPerBlockSize = 40 // Each BlockInfo is approximately this big
  42. indexBatchSize = 1000 // Either way, don't include more files than this
  43. reqValidationTime = time.Hour // How long to cache validation entries for Request messages
  44. reqValidationCacheSize = 1000 // How many entries to aim for in the validation cache size
  45. )
  46. type service interface {
  47. Serve()
  48. Stop()
  49. Jobs() ([]string, []string) // In progress, Queued
  50. BringToFront(string)
  51. DelayScan(d time.Duration)
  52. IndexUpdated() // Remote index was updated notification
  53. setState(state folderState)
  54. setError(err error)
  55. getState() (folderState, time.Time, error)
  56. }
  57. type Model struct {
  58. *suture.Supervisor
  59. cfg *config.Wrapper
  60. db *leveldb.DB
  61. finder *db.BlockFinder
  62. progressEmitter *ProgressEmitter
  63. id protocol.DeviceID
  64. shortID uint64
  65. deviceName string
  66. clientName string
  67. clientVersion string
  68. folderCfgs map[string]config.FolderConfiguration // folder -> cfg
  69. folderFiles map[string]*db.FileSet // folder -> files
  70. folderDevices map[string][]protocol.DeviceID // folder -> deviceIDs
  71. deviceFolders map[protocol.DeviceID][]string // deviceID -> folders
  72. deviceStatRefs map[protocol.DeviceID]*stats.DeviceStatisticsReference // deviceID -> statsRef
  73. folderIgnores map[string]*ignore.Matcher // folder -> matcher object
  74. folderRunners map[string]service // folder -> puller or scanner
  75. folderStatRefs map[string]*stats.FolderStatisticsReference // folder -> statsRef
  76. fmut sync.RWMutex // protects the above
  77. protoConn map[protocol.DeviceID]protocol.Connection
  78. rawConn map[protocol.DeviceID]io.Closer
  79. deviceVer map[protocol.DeviceID]string
  80. pmut sync.RWMutex // protects protoConn and rawConn
  81. addedFolder bool
  82. started bool
  83. reqValidationCache map[string]time.Time // folder / file name => time when confirmed to exist
  84. rvmut sync.RWMutex // protects reqValidationCache
  85. }
  86. var (
  87. SymlinkWarning = stdsync.Once{}
  88. )
  89. // NewModel creates and starts a new model. The model starts in read-only mode,
  90. // where it sends index information to connected peers and responds to requests
  91. // for file data without altering the local folder in any way.
  92. func NewModel(cfg *config.Wrapper, id protocol.DeviceID, deviceName, clientName, clientVersion string, ldb *leveldb.DB) *Model {
  93. m := &Model{
  94. Supervisor: suture.New("model", suture.Spec{
  95. Log: func(line string) {
  96. if debug {
  97. l.Debugln(line)
  98. }
  99. },
  100. }),
  101. cfg: cfg,
  102. db: ldb,
  103. finder: db.NewBlockFinder(ldb, cfg),
  104. progressEmitter: NewProgressEmitter(cfg),
  105. id: id,
  106. shortID: id.Short(),
  107. deviceName: deviceName,
  108. clientName: clientName,
  109. clientVersion: clientVersion,
  110. folderCfgs: make(map[string]config.FolderConfiguration),
  111. folderFiles: make(map[string]*db.FileSet),
  112. folderDevices: make(map[string][]protocol.DeviceID),
  113. deviceFolders: make(map[protocol.DeviceID][]string),
  114. deviceStatRefs: make(map[protocol.DeviceID]*stats.DeviceStatisticsReference),
  115. folderIgnores: make(map[string]*ignore.Matcher),
  116. folderRunners: make(map[string]service),
  117. folderStatRefs: make(map[string]*stats.FolderStatisticsReference),
  118. protoConn: make(map[protocol.DeviceID]protocol.Connection),
  119. rawConn: make(map[protocol.DeviceID]io.Closer),
  120. deviceVer: make(map[protocol.DeviceID]string),
  121. reqValidationCache: make(map[string]time.Time),
  122. fmut: sync.NewRWMutex(),
  123. pmut: sync.NewRWMutex(),
  124. rvmut: sync.NewRWMutex(),
  125. }
  126. if cfg.Options().ProgressUpdateIntervalS > -1 {
  127. go m.progressEmitter.Serve()
  128. }
  129. return m
  130. }
  131. // StartDeadlockDetector starts a deadlock detector on the models locks which
  132. // causes panics in case the locks cannot be acquired in the given timeout
  133. // period.
  134. func (m *Model) StartDeadlockDetector(timeout time.Duration) {
  135. l.Infof("Starting deadlock detector with %v timeout", timeout)
  136. deadlockDetect(m.fmut, timeout)
  137. deadlockDetect(m.pmut, timeout)
  138. }
  139. // StartFolderRW starts read/write processing on the current model. When in
  140. // read/write mode the model will attempt to keep in sync with the cluster by
  141. // pulling needed files from peer devices.
  142. func (m *Model) StartFolderRW(folder string) {
  143. m.fmut.Lock()
  144. cfg, ok := m.folderCfgs[folder]
  145. if !ok {
  146. panic("cannot start nonexistent folder " + folder)
  147. }
  148. _, ok = m.folderRunners[folder]
  149. if ok {
  150. panic("cannot start already running folder " + folder)
  151. }
  152. p := newRWFolder(m, m.shortID, cfg)
  153. m.folderRunners[folder] = p
  154. m.fmut.Unlock()
  155. if len(cfg.Versioning.Type) > 0 {
  156. factory, ok := versioner.Factories[cfg.Versioning.Type]
  157. if !ok {
  158. l.Fatalf("Requested versioning type %q that does not exist", cfg.Versioning.Type)
  159. }
  160. versioner := factory(folder, cfg.Path(), cfg.Versioning.Params)
  161. if service, ok := versioner.(suture.Service); ok {
  162. // The versioner implements the suture.Service interface, so
  163. // expects to be run in the background in addition to being called
  164. // when files are going to be archived.
  165. m.Add(service)
  166. }
  167. p.versioner = versioner
  168. }
  169. go p.Serve()
  170. }
  171. // StartFolderRO starts read only processing on the current model. When in
  172. // read only mode the model will announce files to the cluster but not pull in
  173. // any external changes.
  174. func (m *Model) StartFolderRO(folder string) {
  175. m.fmut.Lock()
  176. cfg, ok := m.folderCfgs[folder]
  177. if !ok {
  178. panic("cannot start nonexistent folder " + folder)
  179. }
  180. _, ok = m.folderRunners[folder]
  181. if ok {
  182. panic("cannot start already running folder " + folder)
  183. }
  184. s := newROFolder(m, folder, time.Duration(cfg.RescanIntervalS)*time.Second)
  185. m.folderRunners[folder] = s
  186. m.fmut.Unlock()
  187. go s.Serve()
  188. }
  189. type ConnectionInfo struct {
  190. protocol.Statistics
  191. Address string
  192. ClientVersion string
  193. }
  194. func (info ConnectionInfo) MarshalJSON() ([]byte, error) {
  195. return json.Marshal(map[string]interface{}{
  196. "at": info.At,
  197. "inBytesTotal": info.InBytesTotal,
  198. "outBytesTotal": info.OutBytesTotal,
  199. "address": info.Address,
  200. "clientVersion": info.ClientVersion,
  201. })
  202. }
  203. // ConnectionStats returns a map with connection statistics for each connected device.
  204. func (m *Model) ConnectionStats() map[string]interface{} {
  205. type remoteAddrer interface {
  206. RemoteAddr() net.Addr
  207. }
  208. m.pmut.RLock()
  209. m.fmut.RLock()
  210. var res = make(map[string]interface{})
  211. conns := make(map[string]ConnectionInfo, len(m.protoConn))
  212. for device, conn := range m.protoConn {
  213. ci := ConnectionInfo{
  214. Statistics: conn.Statistics(),
  215. ClientVersion: m.deviceVer[device],
  216. }
  217. if nc, ok := m.rawConn[device].(remoteAddrer); ok {
  218. ci.Address = nc.RemoteAddr().String()
  219. }
  220. conns[device.String()] = ci
  221. }
  222. res["connections"] = conns
  223. m.fmut.RUnlock()
  224. m.pmut.RUnlock()
  225. in, out := protocol.TotalInOut()
  226. res["total"] = ConnectionInfo{
  227. Statistics: protocol.Statistics{
  228. At: time.Now(),
  229. InBytesTotal: in,
  230. OutBytesTotal: out,
  231. },
  232. }
  233. return res
  234. }
  235. // DeviceStatistics returns statistics about each device
  236. func (m *Model) DeviceStatistics() map[string]stats.DeviceStatistics {
  237. var res = make(map[string]stats.DeviceStatistics)
  238. for id := range m.cfg.Devices() {
  239. res[id.String()] = m.deviceStatRef(id).GetStatistics()
  240. }
  241. return res
  242. }
  243. // FolderStatistics returns statistics about each folder
  244. func (m *Model) FolderStatistics() map[string]stats.FolderStatistics {
  245. var res = make(map[string]stats.FolderStatistics)
  246. for id := range m.cfg.Folders() {
  247. res[id] = m.folderStatRef(id).GetStatistics()
  248. }
  249. return res
  250. }
  251. // Completion returns the completion status, in percent, for the given device
  252. // and folder.
  253. func (m *Model) Completion(device protocol.DeviceID, folder string) float64 {
  254. var tot int64
  255. m.fmut.RLock()
  256. rf, ok := m.folderFiles[folder]
  257. m.fmut.RUnlock()
  258. if !ok {
  259. return 0 // Folder doesn't exist, so we hardly have any of it
  260. }
  261. rf.WithGlobalTruncated(func(f db.FileIntf) bool {
  262. if !f.IsDeleted() {
  263. tot += f.Size()
  264. }
  265. return true
  266. })
  267. if tot == 0 {
  268. return 100 // Folder is empty, so we have all of it
  269. }
  270. var need int64
  271. rf.WithNeedTruncated(device, func(f db.FileIntf) bool {
  272. if !f.IsDeleted() {
  273. need += f.Size()
  274. }
  275. return true
  276. })
  277. res := 100 * (1 - float64(need)/float64(tot))
  278. if debug {
  279. l.Debugf("%v Completion(%s, %q): %f (%d / %d)", m, device, folder, res, need, tot)
  280. }
  281. return res
  282. }
  283. func sizeOf(fs []protocol.FileInfo) (files, deleted int, bytes int64) {
  284. for _, f := range fs {
  285. fs, de, by := sizeOfFile(f)
  286. files += fs
  287. deleted += de
  288. bytes += by
  289. }
  290. return
  291. }
  292. func sizeOfFile(f db.FileIntf) (files, deleted int, bytes int64) {
  293. if !f.IsDeleted() {
  294. files++
  295. } else {
  296. deleted++
  297. }
  298. bytes += f.Size()
  299. return
  300. }
  301. // GlobalSize returns the number of files, deleted files and total bytes for all
  302. // files in the global model.
  303. func (m *Model) GlobalSize(folder string) (nfiles, deleted int, bytes int64) {
  304. m.fmut.RLock()
  305. defer m.fmut.RUnlock()
  306. if rf, ok := m.folderFiles[folder]; ok {
  307. rf.WithGlobalTruncated(func(f db.FileIntf) bool {
  308. fs, de, by := sizeOfFile(f)
  309. nfiles += fs
  310. deleted += de
  311. bytes += by
  312. return true
  313. })
  314. }
  315. return
  316. }
  317. // LocalSize returns the number of files, deleted files and total bytes for all
  318. // files in the local folder.
  319. func (m *Model) LocalSize(folder string) (nfiles, deleted int, bytes int64) {
  320. m.fmut.RLock()
  321. defer m.fmut.RUnlock()
  322. if rf, ok := m.folderFiles[folder]; ok {
  323. rf.WithHaveTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  324. if f.IsInvalid() {
  325. return true
  326. }
  327. fs, de, by := sizeOfFile(f)
  328. nfiles += fs
  329. deleted += de
  330. bytes += by
  331. return true
  332. })
  333. }
  334. return
  335. }
  336. // NeedSize returns the number and total size of currently needed files.
  337. func (m *Model) NeedSize(folder string) (nfiles int, bytes int64) {
  338. m.fmut.RLock()
  339. defer m.fmut.RUnlock()
  340. if rf, ok := m.folderFiles[folder]; ok {
  341. rf.WithNeedTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  342. fs, de, by := sizeOfFile(f)
  343. nfiles += fs + de
  344. bytes += by
  345. return true
  346. })
  347. }
  348. bytes -= m.progressEmitter.BytesCompleted(folder)
  349. if debug {
  350. l.Debugf("%v NeedSize(%q): %d %d", m, folder, nfiles, bytes)
  351. }
  352. return
  353. }
  354. // NeedFolderFiles returns paginated list of currently needed files in
  355. // progress, queued, and to be queued on next puller iteration, as well as the
  356. // total number of files currently needed.
  357. func (m *Model) NeedFolderFiles(folder string, page, perpage int) ([]db.FileInfoTruncated, []db.FileInfoTruncated, []db.FileInfoTruncated, int) {
  358. m.fmut.RLock()
  359. defer m.fmut.RUnlock()
  360. total := 0
  361. rf, ok := m.folderFiles[folder]
  362. if !ok {
  363. return nil, nil, nil, 0
  364. }
  365. var progress, queued, rest []db.FileInfoTruncated
  366. var seen map[string]struct{}
  367. skip := (page - 1) * perpage
  368. get := perpage
  369. runner, ok := m.folderRunners[folder]
  370. if ok {
  371. allProgressNames, allQueuedNames := runner.Jobs()
  372. var progressNames, queuedNames []string
  373. progressNames, skip, get = getChunk(allProgressNames, skip, get)
  374. queuedNames, skip, get = getChunk(allQueuedNames, skip, get)
  375. progress = make([]db.FileInfoTruncated, len(progressNames))
  376. queued = make([]db.FileInfoTruncated, len(queuedNames))
  377. seen = make(map[string]struct{}, len(progressNames)+len(queuedNames))
  378. for i, name := range progressNames {
  379. if f, ok := rf.GetGlobalTruncated(name); ok {
  380. progress[i] = f
  381. seen[name] = struct{}{}
  382. }
  383. }
  384. for i, name := range queuedNames {
  385. if f, ok := rf.GetGlobalTruncated(name); ok {
  386. queued[i] = f
  387. seen[name] = struct{}{}
  388. }
  389. }
  390. }
  391. rest = make([]db.FileInfoTruncated, 0, perpage)
  392. rf.WithNeedTruncated(protocol.LocalDeviceID, func(f db.FileIntf) bool {
  393. total++
  394. if skip > 0 {
  395. skip--
  396. return true
  397. }
  398. if get > 0 {
  399. ft := f.(db.FileInfoTruncated)
  400. if _, ok := seen[ft.Name]; !ok {
  401. rest = append(rest, ft)
  402. get--
  403. }
  404. }
  405. return true
  406. })
  407. return progress, queued, rest, total
  408. }
  409. // Index is called when a new device is connected and we receive their full index.
  410. // Implements the protocol.Model interface.
  411. func (m *Model) Index(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo, flags uint32, options []protocol.Option) {
  412. if flags != 0 {
  413. l.Warnln("protocol error: unknown flags 0x%x in Index message", flags)
  414. return
  415. }
  416. if debug {
  417. l.Debugf("IDX(in): %s %q: %d files", deviceID, folder, len(fs))
  418. }
  419. if !m.folderSharedWith(folder, deviceID) {
  420. events.Default.Log(events.FolderRejected, map[string]string{
  421. "folder": folder,
  422. "device": deviceID.String(),
  423. })
  424. l.Infof("Unexpected folder ID %q sent from device %q; ensure that the folder exists and that this device is selected under \"Share With\" in the folder configuration.", folder, deviceID)
  425. return
  426. }
  427. m.fmut.RLock()
  428. files, ok := m.folderFiles[folder]
  429. runner := m.folderRunners[folder]
  430. m.fmut.RUnlock()
  431. if runner != nil {
  432. // Runner may legitimately not be set if this is the "cleanup" Index
  433. // message at startup.
  434. defer runner.IndexUpdated()
  435. }
  436. if !ok {
  437. l.Fatalf("Index for nonexistant folder %q", folder)
  438. }
  439. for i := 0; i < len(fs); {
  440. if fs[i].Flags&^protocol.FlagsAll != 0 {
  441. if debug {
  442. l.Debugln("dropping update for file with unknown bits set", fs[i])
  443. }
  444. fs[i] = fs[len(fs)-1]
  445. fs = fs[:len(fs)-1]
  446. } else if symlinkInvalid(fs[i].IsSymlink()) {
  447. if debug {
  448. l.Debugln("dropping update for unsupported symlink", fs[i])
  449. }
  450. fs[i] = fs[len(fs)-1]
  451. fs = fs[:len(fs)-1]
  452. } else {
  453. i++
  454. }
  455. }
  456. files.Replace(deviceID, fs)
  457. events.Default.Log(events.RemoteIndexUpdated, map[string]interface{}{
  458. "device": deviceID.String(),
  459. "folder": folder,
  460. "items": len(fs),
  461. "version": files.LocalVersion(deviceID),
  462. })
  463. }
  464. // IndexUpdate is called for incremental updates to connected devices' indexes.
  465. // Implements the protocol.Model interface.
  466. func (m *Model) IndexUpdate(deviceID protocol.DeviceID, folder string, fs []protocol.FileInfo, flags uint32, options []protocol.Option) {
  467. if flags != 0 {
  468. l.Warnln("protocol error: unknown flags 0x%x in IndexUpdate message", flags)
  469. return
  470. }
  471. if debug {
  472. l.Debugf("%v IDXUP(in): %s / %q: %d files", m, deviceID, folder, len(fs))
  473. }
  474. if !m.folderSharedWith(folder, deviceID) {
  475. l.Infof("Update for unexpected folder ID %q sent from device %q; ensure that the folder exists and that this device is selected under \"Share With\" in the folder configuration.", folder, deviceID)
  476. return
  477. }
  478. m.fmut.RLock()
  479. files := m.folderFiles[folder]
  480. runner, ok := m.folderRunners[folder]
  481. m.fmut.RUnlock()
  482. if !ok {
  483. l.Fatalf("IndexUpdate for nonexistant folder %q", folder)
  484. }
  485. for i := 0; i < len(fs); {
  486. if fs[i].Flags&^protocol.FlagsAll != 0 {
  487. if debug {
  488. l.Debugln("dropping update for file with unknown bits set", fs[i])
  489. }
  490. fs[i] = fs[len(fs)-1]
  491. fs = fs[:len(fs)-1]
  492. } else if symlinkInvalid(fs[i].IsSymlink()) {
  493. if debug {
  494. l.Debugln("dropping update for unsupported symlink", fs[i])
  495. }
  496. fs[i] = fs[len(fs)-1]
  497. fs = fs[:len(fs)-1]
  498. } else {
  499. i++
  500. }
  501. }
  502. files.Update(deviceID, fs)
  503. events.Default.Log(events.RemoteIndexUpdated, map[string]interface{}{
  504. "device": deviceID.String(),
  505. "folder": folder,
  506. "items": len(fs),
  507. "version": files.LocalVersion(deviceID),
  508. })
  509. runner.IndexUpdated()
  510. }
  511. func (m *Model) folderSharedWith(folder string, deviceID protocol.DeviceID) bool {
  512. m.fmut.RLock()
  513. defer m.fmut.RUnlock()
  514. for _, nfolder := range m.deviceFolders[deviceID] {
  515. if nfolder == folder {
  516. return true
  517. }
  518. }
  519. return false
  520. }
  521. func (m *Model) ClusterConfig(deviceID protocol.DeviceID, cm protocol.ClusterConfigMessage) {
  522. m.pmut.Lock()
  523. if cm.ClientName == "syncthing" {
  524. m.deviceVer[deviceID] = cm.ClientVersion
  525. } else {
  526. m.deviceVer[deviceID] = cm.ClientName + " " + cm.ClientVersion
  527. }
  528. event := map[string]string{
  529. "id": deviceID.String(),
  530. "clientName": cm.ClientName,
  531. "clientVersion": cm.ClientVersion,
  532. }
  533. if conn, ok := m.rawConn[deviceID].(*tls.Conn); ok {
  534. event["addr"] = conn.RemoteAddr().String()
  535. }
  536. m.pmut.Unlock()
  537. events.Default.Log(events.DeviceConnected, event)
  538. l.Infof(`Device %s client is "%s %s"`, deviceID, cm.ClientName, cm.ClientVersion)
  539. var changed bool
  540. if name := cm.GetOption("name"); name != "" {
  541. l.Infof("Device %s name is %q", deviceID, name)
  542. device, ok := m.cfg.Devices()[deviceID]
  543. if ok && device.Name == "" {
  544. device.Name = name
  545. m.cfg.SetDevice(device)
  546. changed = true
  547. }
  548. }
  549. if m.cfg.Devices()[deviceID].Introducer {
  550. // This device is an introducer. Go through the announced lists of folders
  551. // and devices and add what we are missing.
  552. for _, folder := range cm.Folders {
  553. // If we don't have this folder yet, skip it. Ideally, we'd
  554. // offer up something in the GUI to create the folder, but for the
  555. // moment we only handle folders that we already have.
  556. if _, ok := m.folderDevices[folder.ID]; !ok {
  557. continue
  558. }
  559. nextDevice:
  560. for _, device := range folder.Devices {
  561. var id protocol.DeviceID
  562. copy(id[:], device.ID)
  563. if _, ok := m.cfg.Devices()[id]; !ok {
  564. // The device is currently unknown. Add it to the config.
  565. l.Infof("Adding device %v to config (vouched for by introducer %v)", id, deviceID)
  566. newDeviceCfg := config.DeviceConfiguration{
  567. DeviceID: id,
  568. Compression: m.cfg.Devices()[deviceID].Compression,
  569. Addresses: []string{"dynamic"},
  570. }
  571. // The introducers' introducers are also our introducers.
  572. if device.Flags&protocol.FlagIntroducer != 0 {
  573. l.Infof("Device %v is now also an introducer", id)
  574. newDeviceCfg.Introducer = true
  575. }
  576. m.cfg.SetDevice(newDeviceCfg)
  577. changed = true
  578. }
  579. for _, er := range m.deviceFolders[id] {
  580. if er == folder.ID {
  581. // We already share the folder with this device, so
  582. // nothing to do.
  583. continue nextDevice
  584. }
  585. }
  586. // We don't yet share this folder with this device. Add the device
  587. // to sharing list of the folder.
  588. l.Infof("Adding device %v to share %q (vouched for by introducer %v)", id, folder.ID, deviceID)
  589. m.deviceFolders[id] = append(m.deviceFolders[id], folder.ID)
  590. m.folderDevices[folder.ID] = append(m.folderDevices[folder.ID], id)
  591. folderCfg := m.cfg.Folders()[folder.ID]
  592. folderCfg.Devices = append(folderCfg.Devices, config.FolderDeviceConfiguration{
  593. DeviceID: id,
  594. })
  595. m.cfg.SetFolder(folderCfg)
  596. changed = true
  597. }
  598. }
  599. }
  600. if changed {
  601. m.cfg.Save()
  602. }
  603. }
  604. // Close removes the peer from the model and closes the underlying connection if possible.
  605. // Implements the protocol.Model interface.
  606. func (m *Model) Close(device protocol.DeviceID, err error) {
  607. l.Infof("Connection to %s closed: %v", device, err)
  608. events.Default.Log(events.DeviceDisconnected, map[string]string{
  609. "id": device.String(),
  610. "error": err.Error(),
  611. })
  612. m.pmut.Lock()
  613. m.fmut.RLock()
  614. for _, folder := range m.deviceFolders[device] {
  615. m.folderFiles[folder].Replace(device, nil)
  616. }
  617. m.fmut.RUnlock()
  618. conn, ok := m.rawConn[device]
  619. if ok {
  620. if conn, ok := conn.(*tls.Conn); ok {
  621. // If the underlying connection is a *tls.Conn, Close() does more
  622. // than it says on the tin. Specifically, it sends a TLS alert
  623. // message, which might block forever if the connection is dead
  624. // and we don't have a deadline site.
  625. conn.SetWriteDeadline(time.Now().Add(250 * time.Millisecond))
  626. }
  627. conn.Close()
  628. }
  629. delete(m.protoConn, device)
  630. delete(m.rawConn, device)
  631. delete(m.deviceVer, device)
  632. m.pmut.Unlock()
  633. }
  634. // Request returns the specified data segment by reading it from local disk.
  635. // Implements the protocol.Model interface.
  636. func (m *Model) Request(deviceID protocol.DeviceID, folder, name string, offset int64, size int, hash []byte, flags uint32, options []protocol.Option) ([]byte, error) {
  637. if offset < 0 || size < 0 {
  638. return nil, protocol.ErrNoSuchFile
  639. }
  640. if !m.folderSharedWith(folder, deviceID) {
  641. l.Warnf("Request from %s for file %s in unshared folder %q", deviceID, name, folder)
  642. return nil, protocol.ErrNoSuchFile
  643. }
  644. if flags != 0 {
  645. // We don't currently support or expect any flags.
  646. return nil, fmt.Errorf("protocol error: unknown flags 0x%x in Request message", flags)
  647. }
  648. // Verify that the requested file exists in the local model. We only need
  649. // to validate this file if we haven't done so recently, so we keep a
  650. // cache of successfull results. "Recently" can be quite a long time, as
  651. // we remove validation cache entries when we detect local changes. If
  652. // we're out of sync here and the file actually doesn't exist any more, or
  653. // has shrunk or something, then we'll anyway get a read error that we
  654. // pass on to the other side.
  655. m.rvmut.RLock()
  656. validated := m.reqValidationCache[folder+"/"+name]
  657. m.rvmut.RUnlock()
  658. if time.Since(validated) > reqValidationTime {
  659. m.fmut.RLock()
  660. folderFiles, ok := m.folderFiles[folder]
  661. m.fmut.RUnlock()
  662. if !ok {
  663. l.Warnf("Request from %s for file %s in nonexistent folder %q", deviceID, name, folder)
  664. return nil, protocol.ErrNoSuchFile
  665. }
  666. // This call is really expensive for large files, as we load the full
  667. // block list which may be megabytes and megabytes of data to allocate
  668. // space for, read, and deserialize.
  669. lf, ok := folderFiles.Get(protocol.LocalDeviceID, name)
  670. if !ok {
  671. return nil, protocol.ErrNoSuchFile
  672. }
  673. if lf.IsInvalid() || lf.IsDeleted() {
  674. if debug {
  675. l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d; invalid: %v", m, deviceID, folder, name, offset, size, lf)
  676. }
  677. return nil, protocol.ErrInvalid
  678. }
  679. if offset > lf.Size() {
  680. if debug {
  681. l.Debugf("%v REQ(in; nonexistent): %s: %q o=%d s=%d", m, deviceID, name, offset, size)
  682. }
  683. return nil, protocol.ErrNoSuchFile
  684. }
  685. m.rvmut.Lock()
  686. m.reqValidationCache[folder+"/"+name] = time.Now()
  687. if len(m.reqValidationCache) > reqValidationCacheSize {
  688. // Don't let the cache grow infinitely
  689. for name, validated := range m.reqValidationCache {
  690. if time.Since(validated) > time.Minute {
  691. delete(m.reqValidationCache, name)
  692. }
  693. }
  694. }
  695. m.rvmut.Unlock()
  696. }
  697. if debug && deviceID != protocol.LocalDeviceID {
  698. l.Debugf("%v REQ(in): %s: %q / %q o=%d s=%d", m, deviceID, folder, name, offset, size)
  699. }
  700. m.fmut.RLock()
  701. fn := filepath.Join(m.folderCfgs[folder].Path(), name)
  702. m.fmut.RUnlock()
  703. var reader io.ReaderAt
  704. var err error
  705. if info, err := os.Lstat(fn); err == nil && info.Mode()&os.ModeSymlink != 0 {
  706. target, _, err := symlinks.Read(fn)
  707. if err != nil {
  708. return nil, err
  709. }
  710. reader = strings.NewReader(target)
  711. } else {
  712. // Cannot easily cache fd's because we might need to delete the file
  713. // at any moment.
  714. reader, err = os.Open(fn)
  715. if err != nil {
  716. return nil, err
  717. }
  718. defer reader.(*os.File).Close()
  719. }
  720. buf := make([]byte, size)
  721. _, err = reader.ReadAt(buf, offset)
  722. if err != nil {
  723. return nil, err
  724. }
  725. return buf, nil
  726. }
  727. // ReplaceLocal replaces the local folder index with the given list of files.
  728. func (m *Model) ReplaceLocal(folder string, fs []protocol.FileInfo) {
  729. m.fmut.RLock()
  730. m.folderFiles[folder].ReplaceWithDelete(protocol.LocalDeviceID, fs, m.shortID)
  731. m.fmut.RUnlock()
  732. }
  733. func (m *Model) CurrentFolderFile(folder string, file string) (protocol.FileInfo, bool) {
  734. m.fmut.RLock()
  735. fs, ok := m.folderFiles[folder]
  736. m.fmut.RUnlock()
  737. if !ok {
  738. return protocol.FileInfo{}, false
  739. }
  740. f, ok := fs.Get(protocol.LocalDeviceID, file)
  741. return f, ok
  742. }
  743. func (m *Model) CurrentGlobalFile(folder string, file string) (protocol.FileInfo, bool) {
  744. m.fmut.RLock()
  745. fs, ok := m.folderFiles[folder]
  746. m.fmut.RUnlock()
  747. if !ok {
  748. return protocol.FileInfo{}, false
  749. }
  750. f, ok := fs.GetGlobal(file)
  751. return f, ok
  752. }
  753. type cFiler struct {
  754. m *Model
  755. r string
  756. }
  757. // Implements scanner.CurrentFiler
  758. func (cf cFiler) CurrentFile(file string) (protocol.FileInfo, bool) {
  759. return cf.m.CurrentFolderFile(cf.r, file)
  760. }
  761. // ConnectedTo returns true if we are connected to the named device.
  762. func (m *Model) ConnectedTo(deviceID protocol.DeviceID) bool {
  763. m.pmut.RLock()
  764. _, ok := m.protoConn[deviceID]
  765. m.pmut.RUnlock()
  766. if ok {
  767. m.deviceWasSeen(deviceID)
  768. }
  769. return ok
  770. }
  771. func (m *Model) GetIgnores(folder string) ([]string, []string, error) {
  772. var lines []string
  773. m.fmut.RLock()
  774. cfg, ok := m.folderCfgs[folder]
  775. m.fmut.RUnlock()
  776. if !ok {
  777. return lines, nil, fmt.Errorf("Folder %s does not exist", folder)
  778. }
  779. fd, err := os.Open(filepath.Join(cfg.Path(), ".stignore"))
  780. if err != nil {
  781. if os.IsNotExist(err) {
  782. return lines, nil, nil
  783. }
  784. l.Warnln("Loading .stignore:", err)
  785. return lines, nil, err
  786. }
  787. defer fd.Close()
  788. scanner := bufio.NewScanner(fd)
  789. for scanner.Scan() {
  790. lines = append(lines, strings.TrimSpace(scanner.Text()))
  791. }
  792. m.fmut.RLock()
  793. patterns := m.folderIgnores[folder].Patterns()
  794. m.fmut.RUnlock()
  795. return lines, patterns, nil
  796. }
  797. func (m *Model) SetIgnores(folder string, content []string) error {
  798. cfg, ok := m.folderCfgs[folder]
  799. if !ok {
  800. return fmt.Errorf("Folder %s does not exist", folder)
  801. }
  802. fd, err := ioutil.TempFile(cfg.Path(), ".syncthing.stignore-"+folder)
  803. if err != nil {
  804. l.Warnln("Saving .stignore:", err)
  805. return err
  806. }
  807. defer os.Remove(fd.Name())
  808. for _, line := range content {
  809. _, err = fmt.Fprintln(fd, line)
  810. if err != nil {
  811. l.Warnln("Saving .stignore:", err)
  812. return err
  813. }
  814. }
  815. err = fd.Close()
  816. if err != nil {
  817. l.Warnln("Saving .stignore:", err)
  818. return err
  819. }
  820. file := filepath.Join(cfg.Path(), ".stignore")
  821. err = osutil.Rename(fd.Name(), file)
  822. if err != nil {
  823. l.Warnln("Saving .stignore:", err)
  824. return err
  825. }
  826. return m.ScanFolder(folder)
  827. }
  828. // AddConnection adds a new peer connection to the model. An initial index will
  829. // be sent to the connected peer, thereafter index updates whenever the local
  830. // folder changes.
  831. func (m *Model) AddConnection(rawConn io.Closer, protoConn protocol.Connection) {
  832. deviceID := protoConn.ID()
  833. m.pmut.Lock()
  834. if _, ok := m.protoConn[deviceID]; ok {
  835. panic("add existing device")
  836. }
  837. m.protoConn[deviceID] = protoConn
  838. if _, ok := m.rawConn[deviceID]; ok {
  839. panic("add existing device")
  840. }
  841. m.rawConn[deviceID] = rawConn
  842. cm := m.clusterConfig(deviceID)
  843. protoConn.ClusterConfig(cm)
  844. m.fmut.RLock()
  845. for _, folder := range m.deviceFolders[deviceID] {
  846. fs := m.folderFiles[folder]
  847. go sendIndexes(protoConn, folder, fs, m.folderIgnores[folder])
  848. }
  849. m.fmut.RUnlock()
  850. m.pmut.Unlock()
  851. m.deviceWasSeen(deviceID)
  852. }
  853. func (m *Model) deviceStatRef(deviceID protocol.DeviceID) *stats.DeviceStatisticsReference {
  854. m.fmut.Lock()
  855. defer m.fmut.Unlock()
  856. if sr, ok := m.deviceStatRefs[deviceID]; ok {
  857. return sr
  858. }
  859. sr := stats.NewDeviceStatisticsReference(m.db, deviceID)
  860. m.deviceStatRefs[deviceID] = sr
  861. return sr
  862. }
  863. func (m *Model) deviceWasSeen(deviceID protocol.DeviceID) {
  864. m.deviceStatRef(deviceID).WasSeen()
  865. }
  866. func (m *Model) folderStatRef(folder string) *stats.FolderStatisticsReference {
  867. m.fmut.Lock()
  868. defer m.fmut.Unlock()
  869. sr, ok := m.folderStatRefs[folder]
  870. if !ok {
  871. sr = stats.NewFolderStatisticsReference(m.db, folder)
  872. m.folderStatRefs[folder] = sr
  873. }
  874. return sr
  875. }
  876. func (m *Model) receivedFile(folder, filename string) {
  877. m.folderStatRef(folder).ReceivedFile(filename)
  878. }
  879. func sendIndexes(conn protocol.Connection, folder string, fs *db.FileSet, ignores *ignore.Matcher) {
  880. deviceID := conn.ID()
  881. name := conn.Name()
  882. var err error
  883. if debug {
  884. l.Debugf("sendIndexes for %s-%s/%q starting", deviceID, name, folder)
  885. }
  886. minLocalVer, err := sendIndexTo(true, 0, conn, folder, fs, ignores)
  887. for err == nil {
  888. time.Sleep(5 * time.Second)
  889. if fs.LocalVersion(protocol.LocalDeviceID) <= minLocalVer {
  890. continue
  891. }
  892. minLocalVer, err = sendIndexTo(false, minLocalVer, conn, folder, fs, ignores)
  893. }
  894. if debug {
  895. l.Debugf("sendIndexes for %s-%s/%q exiting: %v", deviceID, name, folder, err)
  896. }
  897. }
  898. func sendIndexTo(initial bool, minLocalVer int64, conn protocol.Connection, folder string, fs *db.FileSet, ignores *ignore.Matcher) (int64, error) {
  899. deviceID := conn.ID()
  900. name := conn.Name()
  901. batch := make([]protocol.FileInfo, 0, indexBatchSize)
  902. currentBatchSize := 0
  903. maxLocalVer := int64(0)
  904. var err error
  905. fs.WithHave(protocol.LocalDeviceID, func(fi db.FileIntf) bool {
  906. f := fi.(protocol.FileInfo)
  907. if f.LocalVersion <= minLocalVer {
  908. return true
  909. }
  910. if f.LocalVersion > maxLocalVer {
  911. maxLocalVer = f.LocalVersion
  912. }
  913. if ignores.Match(f.Name) || symlinkInvalid(f.IsSymlink()) {
  914. if debug {
  915. l.Debugln("not sending update for ignored/unsupported symlink", f)
  916. }
  917. return true
  918. }
  919. if len(batch) == indexBatchSize || currentBatchSize > indexTargetSize {
  920. if initial {
  921. if err = conn.Index(folder, batch, 0, nil); err != nil {
  922. return false
  923. }
  924. if debug {
  925. l.Debugf("sendIndexes for %s-%s/%q: %d files (<%d bytes) (initial index)", deviceID, name, folder, len(batch), currentBatchSize)
  926. }
  927. initial = false
  928. } else {
  929. if err = conn.IndexUpdate(folder, batch, 0, nil); err != nil {
  930. return false
  931. }
  932. if debug {
  933. l.Debugf("sendIndexes for %s-%s/%q: %d files (<%d bytes) (batched update)", deviceID, name, folder, len(batch), currentBatchSize)
  934. }
  935. }
  936. batch = make([]protocol.FileInfo, 0, indexBatchSize)
  937. currentBatchSize = 0
  938. }
  939. batch = append(batch, f)
  940. currentBatchSize += indexPerFileSize + len(f.Blocks)*indexPerBlockSize
  941. return true
  942. })
  943. if initial && err == nil {
  944. err = conn.Index(folder, batch, 0, nil)
  945. if debug && err == nil {
  946. l.Debugf("sendIndexes for %s-%s/%q: %d files (small initial index)", deviceID, name, folder, len(batch))
  947. }
  948. } else if len(batch) > 0 && err == nil {
  949. err = conn.IndexUpdate(folder, batch, 0, nil)
  950. if debug && err == nil {
  951. l.Debugf("sendIndexes for %s-%s/%q: %d files (last batch)", deviceID, name, folder, len(batch))
  952. }
  953. }
  954. return maxLocalVer, err
  955. }
  956. func (m *Model) updateLocals(folder string, fs []protocol.FileInfo) {
  957. m.fmut.RLock()
  958. m.folderFiles[folder].Update(protocol.LocalDeviceID, fs)
  959. m.fmut.RUnlock()
  960. m.rvmut.Lock()
  961. for _, f := range fs {
  962. delete(m.reqValidationCache, folder+"/"+f.Name)
  963. }
  964. m.rvmut.Unlock()
  965. events.Default.Log(events.LocalIndexUpdated, map[string]interface{}{
  966. "folder": folder,
  967. "items": len(fs),
  968. })
  969. }
  970. func (m *Model) requestGlobal(deviceID protocol.DeviceID, folder, name string, offset int64, size int, hash []byte, flags uint32, options []protocol.Option) ([]byte, error) {
  971. m.pmut.RLock()
  972. nc, ok := m.protoConn[deviceID]
  973. m.pmut.RUnlock()
  974. if !ok {
  975. return nil, fmt.Errorf("requestGlobal: no such device: %s", deviceID)
  976. }
  977. if debug {
  978. l.Debugf("%v REQ(out): %s: %q / %q o=%d s=%d h=%x f=%x op=%s", m, deviceID, folder, name, offset, size, hash, flags, options)
  979. }
  980. return nc.Request(folder, name, offset, size, hash, flags, options)
  981. }
  982. func (m *Model) AddFolder(cfg config.FolderConfiguration) {
  983. if m.started {
  984. panic("cannot add folder to started model")
  985. }
  986. if len(cfg.ID) == 0 {
  987. panic("cannot add empty folder id")
  988. }
  989. m.fmut.Lock()
  990. m.folderCfgs[cfg.ID] = cfg
  991. m.folderFiles[cfg.ID] = db.NewFileSet(cfg.ID, m.db)
  992. m.folderDevices[cfg.ID] = make([]protocol.DeviceID, len(cfg.Devices))
  993. for i, device := range cfg.Devices {
  994. m.folderDevices[cfg.ID][i] = device.DeviceID
  995. m.deviceFolders[device.DeviceID] = append(m.deviceFolders[device.DeviceID], cfg.ID)
  996. }
  997. ignores := ignore.New(m.cfg.Options().CacheIgnoredFiles)
  998. _ = ignores.Load(filepath.Join(cfg.Path(), ".stignore")) // Ignore error, there might not be an .stignore
  999. m.folderIgnores[cfg.ID] = ignores
  1000. m.addedFolder = true
  1001. m.fmut.Unlock()
  1002. }
  1003. func (m *Model) ScanFolders() map[string]error {
  1004. m.fmut.RLock()
  1005. folders := make([]string, 0, len(m.folderCfgs))
  1006. for folder := range m.folderCfgs {
  1007. folders = append(folders, folder)
  1008. }
  1009. m.fmut.RUnlock()
  1010. errors := make(map[string]error, len(m.folderCfgs))
  1011. errorsMut := sync.NewMutex()
  1012. wg := sync.NewWaitGroup()
  1013. wg.Add(len(folders))
  1014. for _, folder := range folders {
  1015. folder := folder
  1016. go func() {
  1017. err := m.ScanFolder(folder)
  1018. if err != nil {
  1019. errorsMut.Lock()
  1020. errors[folder] = err
  1021. errorsMut.Unlock()
  1022. // Potentially sets the error twice, once in the scanner just
  1023. // by doing a check, and once here, if the error returned is
  1024. // the same one as returned by CheckFolderHealth, though
  1025. // duplicate set is handled by setError.
  1026. m.fmut.RLock()
  1027. srv := m.folderRunners[folder]
  1028. m.fmut.RUnlock()
  1029. srv.setError(err)
  1030. }
  1031. wg.Done()
  1032. }()
  1033. }
  1034. wg.Wait()
  1035. return errors
  1036. }
  1037. func (m *Model) ScanFolder(folder string) error {
  1038. return m.ScanFolderSubs(folder, nil)
  1039. }
  1040. func (m *Model) ScanFolderSubs(folder string, subs []string) error {
  1041. for i, sub := range subs {
  1042. sub = osutil.NativeFilename(sub)
  1043. if p := filepath.Clean(filepath.Join(folder, sub)); !strings.HasPrefix(p, folder) {
  1044. return errors.New("invalid subpath")
  1045. }
  1046. subs[i] = sub
  1047. }
  1048. m.fmut.Lock()
  1049. fs := m.folderFiles[folder]
  1050. folderCfg := m.folderCfgs[folder]
  1051. ignores := m.folderIgnores[folder]
  1052. runner, ok := m.folderRunners[folder]
  1053. m.fmut.Unlock()
  1054. // Folders are added to folderRunners only when they are started. We can't
  1055. // scan them before they have started, so that's what we need to check for
  1056. // here.
  1057. if !ok {
  1058. return errors.New("no such folder")
  1059. }
  1060. _ = ignores.Load(filepath.Join(folderCfg.Path(), ".stignore")) // Ignore error, there might not be an .stignore
  1061. // Required to make sure that we start indexing at a directory we're already
  1062. // aware off.
  1063. var unifySubs []string
  1064. nextSub:
  1065. for _, sub := range subs {
  1066. for sub != "" {
  1067. if _, ok = fs.Get(protocol.LocalDeviceID, sub); ok {
  1068. break
  1069. }
  1070. sub = filepath.Dir(sub)
  1071. if sub == "." || sub == string(filepath.Separator) {
  1072. sub = ""
  1073. }
  1074. }
  1075. for _, us := range unifySubs {
  1076. if strings.HasPrefix(sub, us) {
  1077. continue nextSub
  1078. }
  1079. }
  1080. unifySubs = append(unifySubs, sub)
  1081. }
  1082. subs = unifySubs
  1083. w := &scanner.Walker{
  1084. Dir: folderCfg.Path(),
  1085. Subs: subs,
  1086. Matcher: ignores,
  1087. BlockSize: protocol.BlockSize,
  1088. TempNamer: defTempNamer,
  1089. TempLifetime: time.Duration(m.cfg.Options().KeepTemporariesH) * time.Hour,
  1090. CurrentFiler: cFiler{m, folder},
  1091. MtimeRepo: db.NewVirtualMtimeRepo(m.db, folderCfg.ID),
  1092. IgnorePerms: folderCfg.IgnorePerms,
  1093. AutoNormalize: folderCfg.AutoNormalize,
  1094. Hashers: m.numHashers(folder),
  1095. ShortID: m.shortID,
  1096. }
  1097. runner.setState(FolderScanning)
  1098. fchan, err := w.Walk()
  1099. if err != nil {
  1100. runner.setError(err)
  1101. return err
  1102. }
  1103. batchSizeFiles := 100
  1104. batchSizeBlocks := 2048 // about 256 MB
  1105. batch := make([]protocol.FileInfo, 0, batchSizeFiles)
  1106. blocksHandled := 0
  1107. for f := range fchan {
  1108. if len(batch) == batchSizeFiles || blocksHandled > batchSizeBlocks {
  1109. if err := m.CheckFolderHealth(folder); err != nil {
  1110. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, err)
  1111. return err
  1112. }
  1113. m.updateLocals(folder, batch)
  1114. batch = batch[:0]
  1115. blocksHandled = 0
  1116. }
  1117. batch = append(batch, f)
  1118. blocksHandled += len(f.Blocks)
  1119. }
  1120. if err := m.CheckFolderHealth(folder); err != nil {
  1121. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, err)
  1122. return err
  1123. } else if len(batch) > 0 {
  1124. m.updateLocals(folder, batch)
  1125. }
  1126. batch = batch[:0]
  1127. // TODO: We should limit the Have scanning to start at sub
  1128. seenPrefix := false
  1129. var iterError error
  1130. fs.WithHaveTruncated(protocol.LocalDeviceID, func(fi db.FileIntf) bool {
  1131. f := fi.(db.FileInfoTruncated)
  1132. hasPrefix := len(subs) == 0
  1133. for _, sub := range subs {
  1134. if strings.HasPrefix(f.Name, sub) {
  1135. hasPrefix = true
  1136. break
  1137. }
  1138. }
  1139. // Return true so that we keep iterating, until we get to the part
  1140. // of the tree we are interested in. Then return false so we stop
  1141. // iterating when we've passed the end of the subtree.
  1142. if !hasPrefix {
  1143. return !seenPrefix
  1144. }
  1145. seenPrefix = true
  1146. if !f.IsDeleted() {
  1147. if f.IsInvalid() {
  1148. return true
  1149. }
  1150. if len(batch) == batchSizeFiles {
  1151. if err := m.CheckFolderHealth(folder); err != nil {
  1152. iterError = err
  1153. return false
  1154. }
  1155. m.updateLocals(folder, batch)
  1156. batch = batch[:0]
  1157. }
  1158. if ignores.Match(f.Name) || symlinkInvalid(f.IsSymlink()) {
  1159. // File has been ignored or an unsupported symlink. Set invalid bit.
  1160. if debug {
  1161. l.Debugln("setting invalid bit on ignored", f)
  1162. }
  1163. nf := protocol.FileInfo{
  1164. Name: f.Name,
  1165. Flags: f.Flags | protocol.FlagInvalid,
  1166. Modified: f.Modified,
  1167. Version: f.Version, // The file is still the same, so don't bump version
  1168. }
  1169. batch = append(batch, nf)
  1170. } else if _, err := osutil.Lstat(filepath.Join(folderCfg.Path(), f.Name)); err != nil {
  1171. // File has been deleted.
  1172. // We don't specifically verify that the error is
  1173. // os.IsNotExist because there is a corner case when a
  1174. // directory is suddenly transformed into a file. When that
  1175. // happens, files that were in the directory (that is now a
  1176. // file) are deleted but will return a confusing error ("not a
  1177. // directory") when we try to Lstat() them.
  1178. nf := protocol.FileInfo{
  1179. Name: f.Name,
  1180. Flags: f.Flags | protocol.FlagDeleted,
  1181. Modified: f.Modified,
  1182. Version: f.Version.Update(m.shortID),
  1183. }
  1184. batch = append(batch, nf)
  1185. }
  1186. }
  1187. return true
  1188. })
  1189. if iterError != nil {
  1190. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, iterError)
  1191. return iterError
  1192. }
  1193. if err := m.CheckFolderHealth(folder); err != nil {
  1194. l.Infof("Stopping folder %s mid-scan due to folder error: %s", folder, err)
  1195. return err
  1196. } else if len(batch) > 0 {
  1197. m.updateLocals(folder, batch)
  1198. }
  1199. runner.setState(FolderIdle)
  1200. return nil
  1201. }
  1202. func (m *Model) DelayScan(folder string, next time.Duration) {
  1203. m.fmut.Lock()
  1204. runner, ok := m.folderRunners[folder]
  1205. m.fmut.Unlock()
  1206. if !ok {
  1207. return
  1208. }
  1209. runner.DelayScan(next)
  1210. }
  1211. // numHashers returns the number of hasher routines to use for a given folder,
  1212. // taking into account configuration and available CPU cores.
  1213. func (m *Model) numHashers(folder string) int {
  1214. m.fmut.Lock()
  1215. folderCfg := m.folderCfgs[folder]
  1216. numFolders := len(m.folderCfgs)
  1217. m.fmut.Unlock()
  1218. if folderCfg.Hashers > 0 {
  1219. // Specific value set in the config, use that.
  1220. return folderCfg.Hashers
  1221. }
  1222. if perFolder := runtime.GOMAXPROCS(-1) / numFolders; perFolder > 0 {
  1223. // We have CPUs to spare, divide them per folder.
  1224. return perFolder
  1225. }
  1226. return 1
  1227. }
  1228. // clusterConfig returns a ClusterConfigMessage that is correct for the given peer device
  1229. func (m *Model) clusterConfig(device protocol.DeviceID) protocol.ClusterConfigMessage {
  1230. cm := protocol.ClusterConfigMessage{
  1231. ClientName: m.clientName,
  1232. ClientVersion: m.clientVersion,
  1233. Options: []protocol.Option{
  1234. {
  1235. Key: "name",
  1236. Value: m.deviceName,
  1237. },
  1238. },
  1239. }
  1240. m.fmut.RLock()
  1241. for _, folder := range m.deviceFolders[device] {
  1242. cr := protocol.Folder{
  1243. ID: folder,
  1244. }
  1245. for _, device := range m.folderDevices[folder] {
  1246. // DeviceID is a value type, but with an underlying array. Copy it
  1247. // so we don't grab aliases to the same array later on in device[:]
  1248. device := device
  1249. // TODO: Set read only bit when relevant
  1250. cn := protocol.Device{
  1251. ID: device[:],
  1252. Flags: protocol.FlagShareTrusted,
  1253. }
  1254. if deviceCfg := m.cfg.Devices()[device]; deviceCfg.Introducer {
  1255. cn.Flags |= protocol.FlagIntroducer
  1256. }
  1257. cr.Devices = append(cr.Devices, cn)
  1258. }
  1259. cm.Folders = append(cm.Folders, cr)
  1260. }
  1261. m.fmut.RUnlock()
  1262. return cm
  1263. }
  1264. func (m *Model) State(folder string) (string, time.Time, error) {
  1265. m.fmut.RLock()
  1266. runner, ok := m.folderRunners[folder]
  1267. m.fmut.RUnlock()
  1268. if !ok {
  1269. // The returned error should be an actual folder error, so returning
  1270. // errors.New("does not exist") or similar here would be
  1271. // inappropriate.
  1272. return "", time.Time{}, nil
  1273. }
  1274. state, changed, err := runner.getState()
  1275. return state.String(), changed, err
  1276. }
  1277. func (m *Model) Override(folder string) {
  1278. m.fmut.RLock()
  1279. fs, ok := m.folderFiles[folder]
  1280. runner := m.folderRunners[folder]
  1281. m.fmut.RUnlock()
  1282. if !ok {
  1283. return
  1284. }
  1285. runner.setState(FolderScanning)
  1286. batch := make([]protocol.FileInfo, 0, indexBatchSize)
  1287. fs.WithNeed(protocol.LocalDeviceID, func(fi db.FileIntf) bool {
  1288. need := fi.(protocol.FileInfo)
  1289. if len(batch) == indexBatchSize {
  1290. fs.Update(protocol.LocalDeviceID, batch)
  1291. batch = batch[:0]
  1292. }
  1293. have, ok := fs.Get(protocol.LocalDeviceID, need.Name)
  1294. if !ok || have.Name != need.Name {
  1295. // We are missing the file
  1296. need.Flags |= protocol.FlagDeleted
  1297. need.Blocks = nil
  1298. need.Version = need.Version.Update(m.shortID)
  1299. } else {
  1300. // We have the file, replace with our version
  1301. have.Version = have.Version.Merge(need.Version).Update(m.shortID)
  1302. need = have
  1303. }
  1304. need.LocalVersion = 0
  1305. batch = append(batch, need)
  1306. return true
  1307. })
  1308. if len(batch) > 0 {
  1309. fs.Update(protocol.LocalDeviceID, batch)
  1310. }
  1311. runner.setState(FolderIdle)
  1312. }
  1313. // CurrentLocalVersion returns the change version for the given folder.
  1314. // This is guaranteed to increment if the contents of the local folder has
  1315. // changed.
  1316. func (m *Model) CurrentLocalVersion(folder string) int64 {
  1317. m.fmut.RLock()
  1318. fs, ok := m.folderFiles[folder]
  1319. m.fmut.RUnlock()
  1320. if !ok {
  1321. // The folder might not exist, since this can be called with a user
  1322. // specified folder name from the REST interface.
  1323. return 0
  1324. }
  1325. return fs.LocalVersion(protocol.LocalDeviceID)
  1326. }
  1327. // RemoteLocalVersion returns the change version for the given folder, as
  1328. // sent by remote peers. This is guaranteed to increment if the contents of
  1329. // the remote or global folder has changed.
  1330. func (m *Model) RemoteLocalVersion(folder string) int64 {
  1331. m.fmut.RLock()
  1332. defer m.fmut.RUnlock()
  1333. fs, ok := m.folderFiles[folder]
  1334. if !ok {
  1335. // The folder might not exist, since this can be called with a user
  1336. // specified folder name from the REST interface.
  1337. return 0
  1338. }
  1339. var ver int64
  1340. for _, n := range m.folderDevices[folder] {
  1341. ver += fs.LocalVersion(n)
  1342. }
  1343. return ver
  1344. }
  1345. func (m *Model) GlobalDirectoryTree(folder, prefix string, levels int, dirsonly bool) map[string]interface{} {
  1346. m.fmut.RLock()
  1347. files, ok := m.folderFiles[folder]
  1348. m.fmut.RUnlock()
  1349. if !ok {
  1350. return nil
  1351. }
  1352. output := make(map[string]interface{})
  1353. sep := string(filepath.Separator)
  1354. prefix = osutil.NativeFilename(prefix)
  1355. if prefix != "" && !strings.HasSuffix(prefix, sep) {
  1356. prefix = prefix + sep
  1357. }
  1358. files.WithPrefixedGlobalTruncated(prefix, func(fi db.FileIntf) bool {
  1359. f := fi.(db.FileInfoTruncated)
  1360. if f.IsInvalid() || f.IsDeleted() || f.Name == prefix {
  1361. return true
  1362. }
  1363. f.Name = strings.Replace(f.Name, prefix, "", 1)
  1364. var dir, base string
  1365. if f.IsDirectory() && !f.IsSymlink() {
  1366. dir = f.Name
  1367. } else {
  1368. dir = filepath.Dir(f.Name)
  1369. base = filepath.Base(f.Name)
  1370. }
  1371. if levels > -1 && strings.Count(f.Name, sep) > levels {
  1372. return true
  1373. }
  1374. last := output
  1375. if dir != "." {
  1376. for _, path := range strings.Split(dir, sep) {
  1377. directory, ok := last[path]
  1378. if !ok {
  1379. newdir := make(map[string]interface{})
  1380. last[path] = newdir
  1381. last = newdir
  1382. } else {
  1383. last = directory.(map[string]interface{})
  1384. }
  1385. }
  1386. }
  1387. if !dirsonly && base != "" {
  1388. last[base] = []interface{}{
  1389. time.Unix(f.Modified, 0), f.Size(),
  1390. }
  1391. }
  1392. return true
  1393. })
  1394. return output
  1395. }
  1396. func (m *Model) Availability(folder, file string) []protocol.DeviceID {
  1397. // Acquire this lock first, as the value returned from foldersFiles can
  1398. // get heavily modified on Close()
  1399. m.pmut.RLock()
  1400. defer m.pmut.RUnlock()
  1401. m.fmut.RLock()
  1402. fs, ok := m.folderFiles[folder]
  1403. m.fmut.RUnlock()
  1404. if !ok {
  1405. return nil
  1406. }
  1407. availableDevices := []protocol.DeviceID{}
  1408. for _, device := range fs.Availability(file) {
  1409. _, ok := m.protoConn[device]
  1410. if ok {
  1411. availableDevices = append(availableDevices, device)
  1412. }
  1413. }
  1414. return availableDevices
  1415. }
  1416. // BringToFront bumps the given files priority in the job queue.
  1417. func (m *Model) BringToFront(folder, file string) {
  1418. m.pmut.RLock()
  1419. defer m.pmut.RUnlock()
  1420. runner, ok := m.folderRunners[folder]
  1421. if ok {
  1422. runner.BringToFront(file)
  1423. }
  1424. }
  1425. // CheckFolderHealth checks the folder for common errors and returns the
  1426. // current folder error, or nil if the folder is healthy.
  1427. func (m *Model) CheckFolderHealth(id string) error {
  1428. folder, ok := m.cfg.Folders()[id]
  1429. if !ok {
  1430. return errors.New("folder does not exist")
  1431. }
  1432. fi, err := os.Stat(folder.Path())
  1433. if m.CurrentLocalVersion(id) > 0 {
  1434. // Safety check. If the cached index contains files but the
  1435. // folder doesn't exist, we have a problem. We would assume
  1436. // that all files have been deleted which might not be the case,
  1437. // so mark it as invalid instead.
  1438. if err != nil || !fi.IsDir() {
  1439. err = errors.New("folder path missing")
  1440. } else if !folder.HasMarker() {
  1441. err = errors.New("folder marker missing")
  1442. }
  1443. } else if os.IsNotExist(err) {
  1444. // If we don't have any files in the index, and the directory
  1445. // doesn't exist, try creating it.
  1446. err = osutil.MkdirAll(folder.Path(), 0700)
  1447. if err == nil {
  1448. err = folder.CreateMarker()
  1449. }
  1450. } else if !folder.HasMarker() {
  1451. // If we don't have any files in the index, and the path does exist
  1452. // but the marker is not there, create it.
  1453. err = folder.CreateMarker()
  1454. }
  1455. m.fmut.RLock()
  1456. runner, runnerExists := m.folderRunners[folder.ID]
  1457. m.fmut.RUnlock()
  1458. var oldErr error
  1459. if runnerExists {
  1460. _, _, oldErr = runner.getState()
  1461. }
  1462. if err != nil {
  1463. if oldErr != nil && oldErr.Error() != err.Error() {
  1464. l.Infof("Folder %q error changed: %q -> %q", folder.ID, oldErr, err)
  1465. } else if oldErr == nil {
  1466. l.Warnf("Stopping folder %q - %v", folder.ID, err)
  1467. }
  1468. if runnerExists {
  1469. runner.setError(err)
  1470. }
  1471. } else if oldErr != nil {
  1472. l.Infof("Folder %q error is cleared, restarting", folder.ID)
  1473. if runnerExists {
  1474. runner.setState(FolderIdle)
  1475. }
  1476. }
  1477. return err
  1478. }
  1479. func (m *Model) ResetFolder(folder string) error {
  1480. for _, f := range db.ListFolders(m.db) {
  1481. if f == folder {
  1482. l.Infof("Cleaning data for folder %q", folder)
  1483. db.DropFolder(m.db, folder)
  1484. return nil
  1485. }
  1486. }
  1487. return fmt.Errorf("Unknown folder %q", folder)
  1488. }
  1489. func (m *Model) String() string {
  1490. return fmt.Sprintf("model@%p", m)
  1491. }
  1492. func (m *Model) VerifyConfiguration(from, to config.Configuration) error {
  1493. return nil
  1494. }
  1495. func (m *Model) CommitConfiguration(from, to config.Configuration) bool {
  1496. // TODO: This should not use reflect, and should take more care to try to handle stuff without restart.
  1497. // Adding, removing or changing folders requires restart
  1498. if !reflect.DeepEqual(from.Folders, to.Folders) {
  1499. return false
  1500. }
  1501. // Removing a device requres restart
  1502. toDevs := make(map[protocol.DeviceID]bool, len(from.Devices))
  1503. for _, dev := range to.Devices {
  1504. toDevs[dev.DeviceID] = true
  1505. }
  1506. for _, dev := range from.Devices {
  1507. if _, ok := toDevs[dev.DeviceID]; !ok {
  1508. return false
  1509. }
  1510. }
  1511. // All of the generic options require restart
  1512. if !reflect.DeepEqual(from.Options, to.Options) {
  1513. return false
  1514. }
  1515. return true
  1516. }
  1517. func symlinkInvalid(isLink bool) bool {
  1518. if !symlinks.Supported && isLink {
  1519. SymlinkWarning.Do(func() {
  1520. l.Warnln("Symlinks are disabled, unsupported or require Administrator privileges. This might cause your folder to appear out of sync.")
  1521. })
  1522. return true
  1523. }
  1524. return false
  1525. }
  1526. // Skips `skip` elements and retrieves up to `get` elements from a given slice.
  1527. // Returns the resulting slice, plus how much elements are left to skip or
  1528. // copy to satisfy the values which were provided, given the slice is not
  1529. // big enough.
  1530. func getChunk(data []string, skip, get int) ([]string, int, int) {
  1531. l := len(data)
  1532. if l <= skip {
  1533. return []string{}, skip - l, get
  1534. } else if l < skip+get {
  1535. return data[skip:l], 0, get - (l - skip)
  1536. }
  1537. return data[skip : skip+get], 0, 0
  1538. }