model.go 44 KB

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