model.go 45 KB

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